/*** test logutil */
use crate::utils::logutil;

/*** test reqwest */
use std::collections::HashMap;

/*** test coolink signal gateway **/
use crate::contract::coolink_sdk::{
    HttpdImpP2PCallBack, HttpdImpP2PContract, P2PRPCServiceCallback, P2PRPCServiceContract,
    SGWebSocketCallBack, SGWebSocketContract,
};
use crate::imp::coolink_sdk_httpd_imp::{FileInLanHttpAddress, HttpSrvContractImp};
use crate::imp::coolink_sdk_p2prpc_imp::{P2PRPCConfig, P2PRPCServiceContractImp};
use crate::imp::coolink_sdk_sg_imp::{InnerSGWebSocketContractImp, SGWebSocketContractImp};
use std::{
    convert::TryInto,
    hash::{Hash, Hasher},
    mem::ManuallyDrop,
    sync::{Arc, RwLock},
    thread,
    thread::sleep,
    thread::JoinHandle,
    time::Duration,
};

use crate::imp::tonicrpc::kcptunnel::tunnel::{
    TUNNEL_ID_BYTEBUF_END_FIX, TUNNEL_ID_FILE_END_FIX, TUNNEL_ID_HTTP_END_FIX,
    TUNNEL_ID_SENSOR_END_FIX,
};
use optional_struct::optional_struct;
use optional_struct::Applyable;
use serde::{Deserialize, Serialize};
use state::Storage;

use crate::model::storage_config::{
    OptionalUserDefaultStorageConfig, UserDefaultStorageConfig, APP_INFO,
    USER_DEFAULT_LOCAL_CONFIG_FILE_NAME,
};
use crate::utils::ids;
use crate::utils::preferences::Preferences;
use std::path::PathBuf;

lazy_static! {
    // state of gateway
    static ref IS_WS_GATWAY_STARTED: Storage<RwLock<bool>> = state::Storage::new();
    // state of p2prpc
    static ref IS_P2PRPC_SRV_STARTED: Storage<RwLock<bool>> = state::Storage::new();
    // state of httpd
    static ref IS_HTTPD_SRV_STARTED: Storage<RwLock<bool>> = state::Storage::new();
    // easyapi global instance
    static ref IS_MANUALLY_EXITED: Storage<RwLock<bool>> = state::Storage::new();
    static ref EASY_API_INSTANCE: Storage<RwLock<CoolinkSDKEasyApi>> = state::Storage::new();

    // unique message tunnel callback
    static ref UNIQUE_MSG_TUNNEL_CB: Storage<RwLock<Arc<Box<dyn IUniqueMessageTunnelListener>>>> = state::Storage::new();
    // unique sensor tunnel callback
    static ref UNIQUE_SENSOR_TUNNEL_CB: Storage<RwLock<Arc<Box<dyn ISensorTunnelListener>>>> = state::Storage::new();
    // unique bytebuf tunnel callback
    static ref UNIQUE_BYTEBUF_TUNNEL_CB: Storage<RwLock<Arc<Box<dyn IByteBufListener>>>> = state::Storage::new();

    static ref LOCAL_SAVED_CLIENT_ID: RwLock<Option<String>> = RwLock::new(None);
    static ref P2PRTCPAIR_MAP: RwLock<HashMap<String, (P2PRPCTunnelPair,TunnelConnectionState)>> = RwLock::new(HashMap::new());

    static ref UNIQUEMSGTUNNEL_MAP: RwLock<HashMap<UniqueTunnel, (P2PRPCTunnelPair,TunnelConnectionState,Arc<Box<dyn IUniqueMessageTunnelListener>>)>> = RwLock::new(HashMap::new());
    static ref UNIQUESENSORTUNNEL_MAP: RwLock<HashMap<UniqueSensorTunnel, (P2PRPCTunnelPair,TunnelConnectionState,Arc<Box<dyn ISensorTunnelListener>>)>> = RwLock::new(HashMap::new());
    static ref UNIQUEBYTEBUFTUNNEL_MAP: RwLock<HashMap<UniqueByteBufTunnel, (P2PRPCTunnelPair,TunnelConnectionState,Arc<Box<dyn IByteBufListener>>)>> = RwLock::new(HashMap::new());

    static ref CONTROL_KIND_MAP: RwLock<HashMap<String, ControlKind>> = RwLock::new(HashMap::new());

    static ref CURRENT_NEW_USER_ID: RwLock<Option<String>> = RwLock::new(None);
    static ref CURRENT_NEW_DEVICE_ID: RwLock<Option<String>> = RwLock::new(None);
}

/// 酷连协议设备控制端/受控端区分枚举
#[derive(Clone, Serialize, Deserialize, Debug, PartialEq)]
pub enum ControlKind {
    CONTROLLING, // 控制端
    CONTROLLED,  // 受控端
}

/// 酷连协议设备连接通道承载数据类型区分枚举
#[derive(Clone, Serialize, Deserialize, Debug, PartialEq)]
pub enum TunnelDataTypeKind {
    LINKED,  //用于字符传输
    FILE,    //用于文件传输
    SENSOR,  //用于密集传感数据传输
    HTTP,    //用于Http流数据传输
    BYTEBUF, //用于音视频流数据传输
}

#[derive(Clone, Serialize, Deserialize, Debug, PartialEq)]
pub enum TunnelConnectionState {
    Unspecified,
    New,
    Checking,
    Connected,
    Completed,
    Failed,
    Disconnected,
    Closed,
}

impl TunnelConnectionState {
    pub fn from_state_string(state: String) -> TunnelConnectionState {
        let mut tunnel_state = TunnelConnectionState::Unspecified;
        if state.eq_ignore_ascii_case(&"New") {
            tunnel_state = TunnelConnectionState::New;
        } else if state.eq_ignore_ascii_case(&"Checking") {
            tunnel_state = TunnelConnectionState::Checking;
        } else if state.eq_ignore_ascii_case(&"Connected") {
            tunnel_state = TunnelConnectionState::Connected;
        } else if state.eq_ignore_ascii_case(&"Completed") {
            tunnel_state = TunnelConnectionState::Completed;
        } else if state.eq_ignore_ascii_case(&"Failed") {
            tunnel_state = TunnelConnectionState::Failed;
        } else if state.eq_ignore_ascii_case(&"Disconnected") {
            tunnel_state = TunnelConnectionState::Disconnected;
        } else if state.eq_ignore_ascii_case(&"Closed") {
            tunnel_state = TunnelConnectionState::Closed;
        }
        tunnel_state
    }
}

/// P2PRPC连接通道属性、类型
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)]
pub struct P2PRPCTunnelPair {
    pub tunnel_id: String,
    pub local_client_id: String,
    pub remote_client_id: String,
    pub control_type: ControlKind,
    pub tunnel_data_type: TunnelDataTypeKind,
    pub is_first_time_inited: bool,
}

impl P2PRPCTunnelPair {
    pub fn new(
        tunnel_id: String,
        local_client_id: String,
        remote_client_id: String,
        control_type: ControlKind,
        tunnel_data_type: TunnelDataTypeKind,
    ) -> Self {
        P2PRPCTunnelPair {
            tunnel_id,
            local_client_id,
            remote_client_id,
            control_type,
            tunnel_data_type,
            is_first_time_inited: true,
        }
    }

    pub fn is_controlling(&self) -> bool {
        if self.control_type == ControlKind::CONTROLLING {
            true
        } else {
            false
        }
    }
}

/// 文本传输通道标识类
#[derive(Debug, Serialize, Deserialize, Clone, Eq, PartialEq, Hash)]
pub struct UniqueTunnel {
    pub tunnel_id: String,
    pub remote_id: String,
}

// impl Hash for UniqueTunnel {
//     fn hash<H: Hasher>(&self, state: &mut H) {
//         self.tunnel_id.hash(state);
//         self.remote_id.hash(state);
//     }
// }

/// sensor传输通道标识类
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq, Hash)]
pub struct UniqueSensorTunnel {
    pub tunnel_id: String,
    pub remote_id: String,
}

/// bytebuf 传输通道标识类
#[derive(Debug, Serialize, Deserialize, Clone, Eq, PartialEq, Hash)]
pub struct UniqueByteBufTunnel {
    pub tunnel_id: String,
    pub remote_id: String,
}

fn set_current_user_id(user_id: String) {
    let mut w = CURRENT_NEW_USER_ID.write().unwrap();
    *w = Some(user_id).to_owned();
}

fn set_current_device_id(device_id: String) {
    let mut w = CURRENT_NEW_DEVICE_ID.write().unwrap();
    *w = Some(device_id.to_owned());
}

pub fn get_current_user_id() -> String {
    let new_user_id_opt = CURRENT_NEW_USER_ID.read().unwrap();
    if (*new_user_id_opt).is_some() {
        (*new_user_id_opt).as_ref().unwrap().clone()
    } else {
        "".to_owned()
    }
}

pub fn get_current_device_id() -> String {
    let new_device_id_opt = CURRENT_NEW_DEVICE_ID.read().unwrap();
    if (*new_device_id_opt).is_some() {
        (*new_device_id_opt).as_ref().unwrap().clone()
    } else {
        "".to_owned()
    }
}

fn get_or_set_local_client_id(is_set: bool, local_client_id: Option<String>) -> Option<String> {
    let res = if is_set {
        let mut local_cid = LOCAL_SAVED_CLIENT_ID.write().unwrap();
        (*local_cid) = local_client_id;

        (*local_cid).clone()
    } else {
        let mut local_cid = LOCAL_SAVED_CLIENT_ID.read().unwrap();
        (*local_cid).clone()
    };

    res
}

fn get_or_set_p2p_rtc_pair_map(
    is_set: bool,
    tunnel_id: String,
    pair: Option<(P2PRPCTunnelPair, TunnelConnectionState)>,
) -> Result<Option<(P2PRPCTunnelPair, TunnelConnectionState)>, anyhow::Error> {
    let res = if is_set {
        let mut map = P2PRTCPAIR_MAP.try_write().map_err(|e| {
            anyhow::anyhow!(
                "get_or_set_p2p_rtc_pair_map P2PRTCPAIR_MAP.try_write() err: {:?} ",
                e
            )
        })?;

        let res = match pair {
            Some(pair_item) => map.insert(tunnel_id.to_owned(), pair_item),
            None => map.remove(&tunnel_id),
        };
        res
    } else {
        match P2PRTCPAIR_MAP
            .try_read()
            .map_err(|e| {
                anyhow::anyhow!(
                    "get_or_set_p2p_rtc_pair_map P2PRTCPAIR_MAP.try_read() err: {:?} ",
                    e
                )
            })?
            .get(&tunnel_id)
        {
            Some(pair_item) => Some(pair_item.clone()),
            None => None,
        }
    };

    Ok(res)
}

fn is_pairmap_contains(tunnel_id: String) -> bool {
    P2PRTCPAIR_MAP.read().unwrap().contains_key(&tunnel_id)
}

fn clear_p2p_rtc_pair_map() -> Result<(), anyhow::Error> {
    let mut map = P2PRTCPAIR_MAP.try_write().map_err(|e| {
        anyhow::anyhow!(
            "clear_p2p_rtc_pair_map P2PRTCPAIR_MAP.try_write() err: {:?} ",
            e
        )
    })?;
    map.clear();
    Ok(())
}

fn get_or_set_control_kind_map(
    is_set: bool,
    tunnel_id: String,
    control_kind: Option<ControlKind>,
) -> Result<Option<ControlKind>, anyhow::Error> {
    let res = if is_set {
        let mut map = CONTROL_KIND_MAP.try_write().map_err(|e| {
            anyhow::anyhow!(
                "get_or_set_control_kind_map CONTROL_KIND_MAP.try_write() err: {:?} ",
                e
            )
        })?;

        let res = match control_kind {
            Some(kind) => map.insert(tunnel_id.to_owned(), kind),
            None => map.remove(&tunnel_id),
        };
        res
    } else {
        match CONTROL_KIND_MAP
            .try_read()
            .map_err(|e| {
                anyhow::anyhow!(
                    "get_or_set_control_kind_map CONTROL_KIND_MAP.try_read() err: {:?} ",
                    e
                )
            })?
            .get(&tunnel_id)
        {
            Some(kind) => Some(kind.clone()),
            None => None,
        }
    };

    Ok(res)
}

fn clear_control_kind_map() -> Result<(), anyhow::Error> {
    let mut map = CONTROL_KIND_MAP.try_write().map_err(|e| {
        anyhow::anyhow!(
            "clear_control_kind_map CONTROL_KIND_MAP.try_write() err: {:?} ",
            e
        )
    })?;
    map.clear();
    Ok(())
}

fn get_or_set_unique_msg_tunnel_map(
    is_set: bool,
    unique_tunnel: UniqueTunnel,
    pair_and_listener: Option<(
        P2PRPCTunnelPair,
        TunnelConnectionState,
        Arc<Box<dyn IUniqueMessageTunnelListener>>,
    )>,
) -> Result<
    Option<(
        P2PRPCTunnelPair,
        TunnelConnectionState,
        Arc<Box<dyn IUniqueMessageTunnelListener>>,
    )>,
    anyhow::Error,
> {
    let res = if is_set {
        let mut map = UNIQUEMSGTUNNEL_MAP.try_write().map_err(|e| {
            anyhow::anyhow!(
                "get_or_set_unique_msg_tunnel_map UNIQUEMSGTUNNEL_MAP.try_write() err: {:?} ",
                e
            )
        })?;

        let res = match pair_and_listener {
            Some(pair_listen_item) => map.insert(unique_tunnel.to_owned(), pair_listen_item),
            None => map.remove(&unique_tunnel),
        };
        res
    } else {
        match UNIQUEMSGTUNNEL_MAP
            .try_read()
            .map_err(|e| {
                anyhow::anyhow!(
                    "get_or_set_unique_msg_tunnel_map UNIQUEMSGTUNNEL_MAP.try_read() err: {:?} ",
                    e
                )
            })?
            .get(&unique_tunnel)
        {
            Some(pair_tuple) => Some(pair_tuple.clone()),
            None => None,
        }
    };

    Ok(res)
}

fn find_unique_msg_tunnel_by_tunnel_id(
    tunnel_id: String,
) -> Result<Option<UniqueTunnel>, anyhow::Error> {
    let map = UNIQUEMSGTUNNEL_MAP.try_read().map_err(|e| {
        anyhow::anyhow!(
            "find_unique_msg_tunnel_by_tunnel_id UNIQUEMSGTUNNEL_MAP.try_read() err: {:?} ",
            e
        )
    })?;
    let mut unique_tunnel_opt = None;
    for (u_tunnel, _) in map.iter() {
        if u_tunnel.tunnel_id.eq_ignore_ascii_case(&tunnel_id) {
            unique_tunnel_opt = Some(u_tunnel.clone());
            break;
        }
    }
    Ok(unique_tunnel_opt)
}

fn find_unique_msg_tunnel_by_remote_c_id(
    remote_id: String,
) -> Result<Vec<Option<UniqueTunnel>>, anyhow::Error> {
    let map = UNIQUEMSGTUNNEL_MAP.try_read().map_err(|e| {
        anyhow::anyhow!(
            "find_unique_msg_tunnel_by_remote_c_id UNIQUEMSGTUNNEL_MAP.try_read() err: {:?} ",
            e
        )
    })?;
    let mut unique_tunnel_opt_vec: Vec<Option<UniqueTunnel>> = vec![];

    for (u_tunnel, _) in map.iter() {
        if u_tunnel.remote_id.eq_ignore_ascii_case(&remote_id) {
            unique_tunnel_opt_vec.push(Some(u_tunnel.clone()))
        }
    }
    Ok(unique_tunnel_opt_vec)
}

fn is_unique_msg_tunnel_map_contains(unique_tunnel: UniqueTunnel) -> bool {
    UNIQUEMSGTUNNEL_MAP
        .read()
        .unwrap()
        .contains_key(&unique_tunnel)
}

fn notify_p2prpc_srv_started(local_client_id: String) -> Result<(), anyhow::Error> {
    let mut map = UNIQUEMSGTUNNEL_MAP.try_write().map_err(|e| {
        anyhow::anyhow!(
            "notify_p2prpc_srv_started UNIQUEMSGTUNNEL_MAP.try_write() err: {:?} ",
            e
        )
    })?;
    if map.len() > 0 {
        for (_, (pair, state, listener)) in map.iter() {
            // UniqueTunnel, (P2PRPCTunnelPair,TunnelConnectionState,Arc<Box<dyn IMessageTunnelListener>>)
            listener.on_p2p_rpc_service_ready(local_client_id.to_owned());
        }
    } else {
        match UNIQUE_MSG_TUNNEL_CB.try_get() {
            Some(listener_instance) => {
                let listener = listener_instance.read().unwrap();
                listener.on_p2p_rpc_service_ready(local_client_id.to_owned());
            }
            None => {
                log::warn!("notify_p2prpc_srv_started failed! please call set_message_type_tunnel_listener first!");
            }
        };
    }

    Ok(())
}

fn notify_httpd_srv_started(local_httpd_host: String) -> Result<(), anyhow::Error> {
    let mut map = UNIQUEMSGTUNNEL_MAP.try_write().map_err(|e| {
        anyhow::anyhow!(
            "notify_httpd_srv_started UNIQUEMSGTUNNEL_MAP.try_write() err: {:?} ",
            e
        )
    })?;
    for (_, (pair, state, listener)) in map.iter() {
        listener.on_http_service_ready(local_httpd_host.to_owned());
    }

    Ok(())
}

fn clear_unique_msg_tunnel_map() -> Result<(), anyhow::Error> {
    let mut map = UNIQUEMSGTUNNEL_MAP.try_write().map_err(|e| {
        anyhow::anyhow!(
            "clear_unique_msg_tunnel_map UNIQUEMSGTUNNEL_MAP.try_write() err: {:?} ",
            e
        )
    })?;
    for (u_tunnel, (pair, state, listener)) in map.drain().take(1) {
        listener.on_disconnected(
            u_tunnel.to_owned(),
            pair.is_controlling(),
            "p2prpc sever destroyed.".to_string(),
        );
    }

    Ok(())
}

fn get_or_set_unique_sensor_tunnel_map(
    is_set: bool,
    unique_sensor_tunnel: UniqueSensorTunnel,
    pair_and_listener: Option<(
        P2PRPCTunnelPair,
        TunnelConnectionState,
        Arc<Box<dyn ISensorTunnelListener>>,
    )>,
) -> Result<
    Option<(
        P2PRPCTunnelPair,
        TunnelConnectionState,
        Arc<Box<dyn ISensorTunnelListener>>,
    )>,
    anyhow::Error,
> {
    let res = if is_set {
        let mut map = UNIQUESENSORTUNNEL_MAP.try_write().map_err(|e| {
            anyhow::anyhow!(
                "get_or_set_unique_sensor_tunnel_map UNIQUESENSORTUNNEL_MAP.try_write() err: {:?} ",
                e
            )
        })?;

        let res = match pair_and_listener {
            Some(pair_listen_item) => map.insert(unique_sensor_tunnel.to_owned(), pair_listen_item),
            None => map.remove(&unique_sensor_tunnel),
        };
        res
    } else {
        match UNIQUESENSORTUNNEL_MAP
            .try_read()
            .map_err(|e| anyhow::anyhow!("get_or_set_unique_sensor_tunnel_map UNIQUESENSORTUNNEL_MAP.try_read() err: {:?} ", e))?
            .get(&unique_sensor_tunnel) {
            Some(pair_tuple) => Some(pair_tuple.clone()),
            None => None,
        }
    };

    Ok(res)
}

fn find_unique_sensor_tunnel_by_tunnel_id(
    tunnel_id: String,
) -> Result<Option<UniqueSensorTunnel>, anyhow::Error> {
    let map = UNIQUESENSORTUNNEL_MAP.try_read().map_err(|e| {
        anyhow::anyhow!(
            "find_unique_sensor_tunnel_by_tunnel_id UNIQUESENSORTUNNEL_MAP.try_read() err: {:?} ",
            e
        )
    })?;
    let mut unique_tunnel_opt = None;
    for (u_tunnel, _) in map.iter() {
        if u_tunnel.tunnel_id.eq_ignore_ascii_case(&tunnel_id) {
            unique_tunnel_opt = Some(u_tunnel.clone());
            break;
        }
    }
    Ok(unique_tunnel_opt)
}

fn find_unique_sensor_tunnel_by_remote_c_id(
    remote_id: String,
) -> Result<Vec<Option<UniqueSensorTunnel>>, anyhow::Error> {
    let map = UNIQUESENSORTUNNEL_MAP.try_read().map_err(|e| {
        anyhow::anyhow!(
            "find_unique_sensor_tunnel_by_remote_c_id UNIQUESENSORTUNNEL_MAP.try_read() err: {:?} ",
            e
        )
    })?;
    let mut unique_tunnel_opt_vec: Vec<Option<UniqueSensorTunnel>> = vec![];

    for (u_tunnel, _) in map.iter() {
        if u_tunnel.remote_id.eq_ignore_ascii_case(&remote_id) {
            unique_tunnel_opt_vec.push(Some(u_tunnel.clone()))
        }
    }
    Ok(unique_tunnel_opt_vec)
}

fn is_unique_sensor_tunnel_map_contains(unique_sensor_tunnel: UniqueSensorTunnel) -> bool {
    UNIQUESENSORTUNNEL_MAP
        .read()
        .unwrap()
        .contains_key(&unique_sensor_tunnel)
}

fn clear_unique_sensor_tunnel_map() -> Result<(), anyhow::Error> {
    let mut map = UNIQUESENSORTUNNEL_MAP.try_write().map_err(|e| {
        anyhow::anyhow!(
            "clear_unique_sensor_tunnel_map UNIQUESENSORTUNNEL_MAP.try_write() err: {:?} ",
            e
        )
    })?;
    for (u_tunnel, (pair, state, listener)) in map.drain().take(1) {
        listener.on_closed(
            u_tunnel.to_owned(),
            pair.is_controlling(),
            "p2prpc sever destroyed.".to_string(),
        );
    }

    Ok(())
}

fn get_or_set_unique_bytebuf_tunnel_map(
    is_set: bool,
    unique_bytebuf_tunnel: UniqueByteBufTunnel,
    pair_and_listener: Option<(
        P2PRPCTunnelPair,
        TunnelConnectionState,
        Arc<Box<dyn IByteBufListener>>,
    )>,
) -> Result<
    Option<(
        P2PRPCTunnelPair,
        TunnelConnectionState,
        Arc<Box<dyn IByteBufListener>>,
    )>,
    anyhow::Error,
> {
    let res = if is_set {
        let mut map = UNIQUEBYTEBUFTUNNEL_MAP
            .try_write()
            .map_err(|e| anyhow::anyhow!("get_or_set_unique_bytebuf_tunnel_map UNIQUEBYTEBUFTUNNEL_MAP.try_write() err: {:?} ", e))?;

        let res = match pair_and_listener {
            Some(pair_listen_item) => {
                map.insert(unique_bytebuf_tunnel.to_owned(), pair_listen_item)
            }
            None => map.remove(&unique_bytebuf_tunnel),
        };
        res
    } else {
        match UNIQUEBYTEBUFTUNNEL_MAP
            .try_read()
            .map_err(|e| anyhow::anyhow!("get_or_set_unique_bytebuf_tunnel_map UNIQUEBYTEBUFTUNNEL_MAP.try_read() err: {:?} ", e))?
            .get(&unique_bytebuf_tunnel)
        {
            Some(pair_tuple) => Some(pair_tuple.clone()),
            None => None,
        }
    };

    Ok(res)
}

fn find_unique_bytebuf_tunnel_by_tunnel_id(
    tunnel_id: String,
) -> Result<Option<UniqueByteBufTunnel>, anyhow::Error> {
    let map = UNIQUEBYTEBUFTUNNEL_MAP.try_read().map_err(|e| {
        anyhow::anyhow!(
            "find_unique_bytebuf_tunnel_by_tunnel_id UNIQUEBYTEBUFTUNNEL_MAP.try_read() err: {:?} ",
            e
        )
    })?;
    let mut unique_tunnel_opt = None;
    for (u_tunnel, _) in map.iter() {
        if u_tunnel.tunnel_id.eq_ignore_ascii_case(&tunnel_id) {
            unique_tunnel_opt = Some(u_tunnel.clone());
            break;
        }
    }
    Ok(unique_tunnel_opt)
}

fn find_unique_bytebuf_tunnel_by_remote_c_id(
    remote_id: String,
) -> Result<Vec<Option<UniqueByteBufTunnel>>, anyhow::Error> {
    let map = UNIQUEBYTEBUFTUNNEL_MAP.try_read().map_err(|e| anyhow::anyhow!("find_unique_bytebuf_tunnel_by_remote_c_id UNIQUEBYTEBUFTUNNEL_MAP.try_read() err: {:?} ", e))?;
    let mut unique_tunnel_opt_vec: Vec<Option<UniqueByteBufTunnel>> = vec![];

    for (u_tunnel, _) in map.iter() {
        if u_tunnel.remote_id.eq_ignore_ascii_case(&remote_id) {
            unique_tunnel_opt_vec.push(Some(u_tunnel.clone()))
        }
    }
    Ok(unique_tunnel_opt_vec)
}

fn is_unique_bytebuf_tunnel_map_contains(unique_bytebuf_tunnel: UniqueByteBufTunnel) -> bool {
    UNIQUEBYTEBUFTUNNEL_MAP
        .read()
        .unwrap()
        .contains_key(&unique_bytebuf_tunnel)
}

fn clear_unique_bytebuf_tunnel_map() -> Result<(), anyhow::Error> {
    let mut map = UNIQUEBYTEBUFTUNNEL_MAP.try_write().map_err(|e| {
        anyhow::anyhow!(
            "clear_unique_bytebuf_tunnel_map UNIQUESENSORTUNNEL_MAP.try_write() err: {:?} ",
            e
        )
    })?;

    for (u_tunnel, (pair, state, listener)) in map.drain().take(1) {
        listener.on_closed(
            u_tunnel.to_owned(),
            pair.is_controlling(),
            "p2prpc sever destroyed.".to_string(),
        );
    }

    Ok(())
}

fn delete_unique_msg_tunnel_map(tunnel_id: String) -> Result<(), anyhow::Error> {
    if let Ok(unique_msg_tunnel_opt) = find_unique_msg_tunnel_by_tunnel_id(tunnel_id.clone()) {
        log::info!("delete_unique_msg_tunnel_map find_unique_msg_tunnel_by_tunnel_id success.");
        if let Some(unique_msg_tunnel) = unique_msg_tunnel_opt {
            // if find messge tunnel already!
            let ((pair, state, listener)) =
                get_or_set_unique_msg_tunnel_map(false, unique_msg_tunnel.clone(), None)
                    .unwrap()
                    .unwrap();
            if pair.is_first_time_inited {
                listener.on_connect_failed(
                    unique_msg_tunnel.clone(),
                    pair.is_controlling(),
                    "p2prpc tunnel disconnected.".to_string(),
                );
            } else {
                listener.on_disconnected(
                    unique_msg_tunnel.clone(),
                    pair.is_controlling(),
                    "p2prpc tunnel disconnected.".to_string(),
                );
            }
            get_or_set_unique_msg_tunnel_map(true, unique_msg_tunnel.to_owned(), None);
        } else {
            // is controlled type
            log::warn!(
                "delete_unique_msg_tunnel_map not find tunnel_id :{:?}",
                tunnel_id
            );
        }
    };
    Ok(())
}

fn delete_unique_sensor_tunnel_map(tunnel_id: String) -> Result<(), anyhow::Error> {
    if let Ok(unique_sensor_tunnel_opt) = find_unique_sensor_tunnel_by_tunnel_id(tunnel_id.clone())
    {
        log::info!(
            "delete_unique_sensor_tunnel_map find_unique_sensor_tunnel_by_tunnel_id success."
        );
        if let Some(unique_sensor_tunnel) = unique_sensor_tunnel_opt {
            // if find messge tunnel already!
            let ((pair, state, listener)) =
                get_or_set_unique_sensor_tunnel_map(false, unique_sensor_tunnel.clone(), None)
                    .unwrap()
                    .unwrap();
            if pair.is_first_time_inited {
                listener.on_open_failed(
                    unique_sensor_tunnel.clone(),
                    pair.is_controlling(),
                    "p2prpc tunnel disconnected.".to_string(),
                )
            } else {
                listener.on_closed(
                    unique_sensor_tunnel.clone(),
                    pair.is_controlling(),
                    "p2prpc tunnel disconnected.".to_string(),
                );
            }
            get_or_set_unique_sensor_tunnel_map(true, unique_sensor_tunnel.to_owned(), None);
        } else {
            // is controlled type
            log::warn!(
                "delete_unique_sensor_tunnel_map not find tunnel_id :{:?}",
                tunnel_id
            );
        }
    };
    Ok(())
}

fn delete_unique_bytebuf_tunnel_map(tunnel_id: String) -> Result<(), anyhow::Error> {
    if let Ok(unique_bytebuf_tunnel_opt) =
        find_unique_bytebuf_tunnel_by_tunnel_id(tunnel_id.clone())
    {
        log::info!(
            "delete_unique_bytebuf_tunnel_map find_unique_bytebuf_tunnel_by_tunnel_id success."
        );
        if let Some(unique_bytebuf_tunnel) = unique_bytebuf_tunnel_opt {
            // if find messge tunnel already!
            let ((pair, state, listener)) =
                get_or_set_unique_bytebuf_tunnel_map(false, unique_bytebuf_tunnel.clone(), None)
                    .unwrap()
                    .unwrap();
            if pair.is_first_time_inited {
                listener.on_open_failed(
                    unique_bytebuf_tunnel.clone(),
                    pair.is_controlling(),
                    "p2prpc tunnel disconnected.".to_string(),
                )
            } else {
                listener.on_closed(
                    unique_bytebuf_tunnel.clone(),
                    pair.is_controlling(),
                    "p2prpc tunnel disconnected.".to_string(),
                );
            }
            get_or_set_unique_bytebuf_tunnel_map(true, unique_bytebuf_tunnel.to_owned(), None);
        } else {
            // is controlled type
            log::warn!(
                "delete_unique_bytebuf_tunnel_map not find tunnel_id :{:?}",
                tunnel_id
            );
        }
    };
    Ok(())
}

fn refresh_pair_map_state(
    tunnel_id: String,
    local_client_id: String,
    remote_client_id: String,
    state: String,
    state_msg: String,
) -> Result<(), anyhow::Error> {
    if let Some((pair, _)) = get_or_set_p2p_rtc_pair_map(false, tunnel_id.clone(), None)? {
        log::info!("refresh_pair_map_state tunnel_id ={},local_client_id ={},remote_client_id={},state={},state_msg={}", tunnel_id, local_client_id,remote_client_id,state, state_msg);
        let new_state = TunnelConnectionState::from_state_string(state);
        match new_state {
            TunnelConnectionState::Failed | TunnelConnectionState::Closed => {
                get_or_set_p2p_rtc_pair_map(true, tunnel_id.clone(), None);
            }
            _ => {
                get_or_set_p2p_rtc_pair_map(
                    true,
                    tunnel_id.clone(),
                    Some((pair.clone(), new_state)),
                );
            }
        };
    } else {
        log::warn!("refresh_pair_map_state but not managed : tunnel_id ={},local_client_id ={},remote_client_id={}", tunnel_id, local_client_id,remote_client_id);
        let new_state = TunnelConnectionState::from_state_string(state);
        let tunnel_data_type = tunnel_data_type_kind(tunnel_id.clone());
        if let Some(control_type) =
            get_or_set_control_kind_map(false, tunnel_id.clone(), None).unwrap()
        {
            get_or_set_p2p_rtc_pair_map(
                true,
                tunnel_id.clone(),
                Some((
                    P2PRPCTunnelPair::new(
                        tunnel_id.clone(),
                        local_client_id,
                        remote_client_id,
                        control_type,
                        tunnel_data_type,
                    ),
                    new_state,
                )),
            );
        }
    }
    Ok(())
}

fn refresh_unique_msg_map_state(
    tunnel_id: String,
    local_client_id: String,
    remote_client_id: String,
    state: String,
    state_msg: String,
) -> Result<(), anyhow::Error> {
    if let Ok(unique_msg_tunnel_opt) = find_unique_msg_tunnel_by_tunnel_id(tunnel_id.clone()) {
        if let Some(unique_msg_tunnel) = unique_msg_tunnel_opt {
            // if find messge tunnel already!
            let ((pair, _, listener)) =
                get_or_set_unique_msg_tunnel_map(false, unique_msg_tunnel.clone(), None)
                    .unwrap()
                    .unwrap();
            log::info!("refresh_unique_msg_map_state tunnel_id ={},local_client_id ={},remote_client_id={},state={},state_msg={}", tunnel_id, local_client_id,remote_client_id,state, state_msg);
            let new_state = TunnelConnectionState::from_state_string(state);
            match new_state {
                TunnelConnectionState::New
                | TunnelConnectionState::Checking
                | TunnelConnectionState::Unspecified => {
                    listener
                        .on_connect_unstable(unique_msg_tunnel.to_owned(), pair.is_controlling());
                }
                TunnelConnectionState::Connected => {
                    log::info!("refresh_unique_msg_map_state select message tunnel as duplex.");
                    P2PRPCServiceContractImp::select_text_tunnel_duplex(tunnel_id.clone());
                }
                TunnelConnectionState::Disconnected => {
                    listener.on_pre_disconnect(unique_msg_tunnel.clone(), pair.is_controlling());
                    // Tips: quick close tunnel
                    P2PRPCServiceContractImp::drop_p2prpc_tunnel(tunnel_id.clone());
                }
                _ => {
                    log::info!("refresh_unique_msg_map_state.");
                }
            }
            get_or_set_unique_msg_tunnel_map(
                true,
                unique_msg_tunnel.to_owned(),
                Some((pair.to_owned(), new_state.to_owned(), listener.to_owned())),
            );
        } else {
            // is controlled type
            log::warn!(
                "refresh_unique_msg_map_state not find tunnel_id :{:?}",
                tunnel_id
            );
        }
    };
    Ok(())
}

fn refresh_unique_sensor_map_state(
    tunnel_id: String,
    local_client_id: String,
    remote_client_id: String,
    state: String,
    state_msg: String,
) -> Result<(), anyhow::Error> {
    if let Ok(unique_sensor_tunnel_opt) = find_unique_sensor_tunnel_by_tunnel_id(tunnel_id.clone())
    {
        if let Some(unique_sensor_tunnel) = unique_sensor_tunnel_opt {
            // if find messge tunnel already!
            let ((pair, _, listener)) =
                get_or_set_unique_sensor_tunnel_map(false, unique_sensor_tunnel.clone(), None)
                    .unwrap()
                    .unwrap();
            log::info!("refresh_unique_sensor_map_state tunnel_id ={},local_client_id ={},remote_client_id={},state={},state_msg={}", tunnel_id, local_client_id,remote_client_id,state, state_msg);
            let new_state = TunnelConnectionState::from_state_string(state);
            match new_state {
                TunnelConnectionState::New
                | TunnelConnectionState::Checking
                | TunnelConnectionState::Unspecified => {
                    listener.on_open_tunnel_unstable(
                        unique_sensor_tunnel.to_owned(),
                        pair.is_controlling(),
                    );
                }
                TunnelConnectionState::Connected => {
                    log::info!("refresh_unique_sensor_map_state select sensor tunnel as duplex.");
                    P2PRPCServiceContractImp::select_text_tunnel_duplex(tunnel_id.clone());
                }
                TunnelConnectionState::Disconnected => {
                    listener.on_pre_closed(unique_sensor_tunnel.clone(), pair.is_controlling());
                    // Tips: quick close tunnel
                    P2PRPCServiceContractImp::drop_p2prpc_tunnel(tunnel_id.clone());
                }
                _ => {
                    log::info!("refresh_unique_sensor_map_state ");
                }
            }
            get_or_set_unique_sensor_tunnel_map(
                true,
                unique_sensor_tunnel.to_owned(),
                Some((pair.to_owned(), new_state, listener.to_owned())),
            );
        } else {
            // is controlled type
            log::warn!(
                "refresh_unique_sensor_map_state not find tunnel_id :{:?}",
                tunnel_id
            );
        }
    };
    Ok(())
}

fn refresh_unique_bytebuf_map_state(
    tunnel_id: String,
    local_client_id: String,
    remote_client_id: String,
    state: String,
    state_msg: String,
) -> Result<(), anyhow::Error> {
    if let Ok(unique_bytebuf_tunnel_opt) =
        find_unique_bytebuf_tunnel_by_tunnel_id(tunnel_id.clone())
    {
        if let Some(unique_bytebuf_tunnel) = unique_bytebuf_tunnel_opt {
            // if find messge tunnel already!
            let ((pair, _, listener)) =
                get_or_set_unique_bytebuf_tunnel_map(false, unique_bytebuf_tunnel.clone(), None)
                    .unwrap()
                    .unwrap();
            log::info!("refresh_unique_bytebuf_map_state tunnel_id ={},local_client_id ={},remote_client_id={},state={},state_msg={}", tunnel_id, local_client_id,remote_client_id,state, state_msg);
            let new_state = TunnelConnectionState::from_state_string(state);
            match new_state {
                TunnelConnectionState::New
                | TunnelConnectionState::Checking
                | TunnelConnectionState::Unspecified => {
                    listener.on_open_tunnel_unstable(
                        unique_bytebuf_tunnel.to_owned(),
                        pair.is_controlling(),
                    );
                }
                TunnelConnectionState::Connected => {
                    log::info!("refresh_unique_bytebuf_map_state select bytebuf tunnel as duplex.");
                    P2PRPCServiceContractImp::select_bytebuf_tunnel_duplex(tunnel_id.clone());
                }
                TunnelConnectionState::Disconnected => {
                    listener.on_pre_closed(unique_bytebuf_tunnel.clone(), pair.is_controlling());
                    // Tips: quick close tunnel
                    P2PRPCServiceContractImp::drop_p2prpc_tunnel(tunnel_id.clone());
                }
                _ => {
                    log::info!("refresh_unique_bytebuf_map_state ");
                }
            }
            get_or_set_unique_bytebuf_tunnel_map(
                true,
                unique_bytebuf_tunnel.to_owned(),
                Some((pair.to_owned(), new_state, listener.to_owned())),
            );
        } else {
            // is controlled type
            log::warn!(
                "refresh_unique_bytebuf_map_state not find tunnel_id :{:?}",
                tunnel_id
            );
        }
    };
    Ok(())
}

fn tunnel_data_type_kind(tunnel_id: String) -> TunnelDataTypeKind {
    let mut data_type_kind = TunnelDataTypeKind::LINKED;
    if tunnel_id.ends_with(TUNNEL_ID_SENSOR_END_FIX) {
        data_type_kind = TunnelDataTypeKind::SENSOR;
    } else if tunnel_id.ends_with(TUNNEL_ID_BYTEBUF_END_FIX) {
        data_type_kind = TunnelDataTypeKind::BYTEBUF;
    } else if tunnel_id.ends_with(TUNNEL_ID_HTTP_END_FIX) {
        data_type_kind = TunnelDataTypeKind::HTTP;
    } else if tunnel_id.ends_with(TUNNEL_ID_FILE_END_FIX) {
        data_type_kind = TunnelDataTypeKind::FILE;
    }
    data_type_kind
}

pub fn set_websocketsrv_started(start_flag: bool) -> bool {
    let ok = if let Some(ws_flag) = IS_WS_GATWAY_STARTED.try_get() {
        if let Ok(mut ws_flag) = ws_flag.try_write() {
            // connect state can write
            if *ws_flag != start_flag {
                *ws_flag = start_flag;
                true
            } else {
                false
            }
        } else {
            false
        }
    } else {
        // set new connect_state
        IS_WS_GATWAY_STARTED.set(RwLock::new(start_flag));
        true
    };
    return ok;
}

pub fn is_websocketsrv_started() -> bool {
    let is_started = if let Some(is_started) = IS_WS_GATWAY_STARTED.try_get() {
        if let Ok(is_started) = is_started.try_read() {
            *is_started
        } else {
            false
        }
    } else {
        false
    };
    is_started
}

pub fn set_p2pprc_started(start_flag: bool) -> bool {
    let ok = if let Some(p2p_flag) = IS_P2PRPC_SRV_STARTED.try_get() {
        if let Ok(mut p2p_flag) = p2p_flag.try_write() {
            // connect state can write
            if *p2p_flag != start_flag {
                *p2p_flag = start_flag;
                true
            } else {
                false
            }
        } else {
            false
        }
    } else {
        // set new connect_state
        IS_P2PRPC_SRV_STARTED.set(RwLock::new(start_flag));
        true
    };
    return ok;
}

pub fn is_p2pprc_started() -> bool {
    let is_started = if let Some(is_started) = IS_P2PRPC_SRV_STARTED.try_get() {
        if let Ok(is_started) = is_started.try_read() {
            *is_started
        } else {
            false
        }
    } else {
        false
    };
    is_started
}

pub fn set_httpd_srv_started(start_flag: bool) -> bool {
    let ok = if let Some(httpd_flag) = IS_HTTPD_SRV_STARTED.try_get() {
        if let Ok(mut httpd_flag) = httpd_flag.try_write() {
            // connect state can write
            if *httpd_flag != start_flag {
                *httpd_flag = start_flag;
                true
            } else {
                false
            }
        } else {
            false
        }
    } else {
        // set new connect_state
        IS_HTTPD_SRV_STARTED.set(RwLock::new(start_flag));
        true
    };
    return ok;
}

pub fn is_httpd_srv_started() -> bool {
    let is_started = if let Some(is_started) = IS_HTTPD_SRV_STARTED.try_get() {
        if let Ok(is_started) = is_started.try_read() {
            *is_started
        } else {
            false
        }
    } else {
        false
    };
    is_started
}

pub fn is_manually_exited() -> bool {
    let is_started = if let Some(is_started) = IS_MANUALLY_EXITED.try_get() {
        if let Ok(is_started) = is_started.try_read() {
            *is_started
        } else {
            false
        }
    } else {
        false
    };
    is_started
}

/// 调用sendMessage是否成功的回调
pub trait ISendMessageResult: Sync + Send {
    fn on_result(&self, success: bool, msg_id: String, err_msg: String);
}

/// 调用sendSensor是否成功的回调
pub trait ISendSensorResult: Sync + Send {
    fn on_result(&self, success: bool, msg_id: String, err_msg: String);
}

/// 调用sendByteBuf是否成功的回调
pub trait ISendByteBufResult: Sync + Send {
    fn on_result(&self, success: bool, msg_id: String, err_msg: String);
}

/// 收到P2P通道回来的消息的回调
pub trait IMessageTunnelListener: Sync + Send {
    /// 调用connectAsControlling，成功或者失败会有回调
    fn on_connect_failed(&self, tunnel: UniqueTunnel, is_controlling: bool, err_reason: String);
    /// 代表有一个连接已经建立成功，is_controlling代表是发起连接的一端或者是被连接的一端
    /// 此时全双工的消息发送通道还未建立好，需要等onReadyToSend
    fn on_connected(&self, tunnel: UniqueTunnel, is_controlling: bool);
    /// 代表了可以发送消息了,
    fn on_ready_to_send(&self, tunnel: UniqueTunnel, ready_to_send: bool);
    /// 因为网络原因导致连接不稳定，内部正在进行重连，重连好了会调用onReadyToSend。
    fn on_connect_unstable(&self, tunnel: UniqueTunnel, is_controlling: bool);
    /// 该状态代表即将发生端开的状态，该状态回调之后，接下去会回调onConnected或者onDisconnected（耗时长）。
    /// 建议收到该消息后，重新完成P2PRPC通道的重置，能够加快连接
    fn on_pre_disconnect(&self, tunnel: UniqueTunnel, is_controlling: bool);
    /// 连接断开了
    fn on_disconnected(&self, tunnel: UniqueTunnel, is_controlling: bool, err_reason: String);
    /// 收到了文本连接通道的消息
    fn on_received(
        &self,
        tunnel: UniqueTunnel,
        is_controlling: bool,
        message_id: String,
        event: String,
        message: String,
    );
}

/// 每一对设备之间的连接都有唯一一个连接和通道，数据回调也在这个通道中
pub trait IUniqueMessageTunnelListener: IMessageTunnelListener {
    /// P2PRPC 服务准备完成
    fn on_p2p_rpc_service_ready(&self, local_client_id: String);
    /// Httpd 服务准备完成
    fn on_http_service_ready(&self, host: String);
}

/// 每一对设备之间的连接都有唯一一个传感数据通道，数据回调也在这个通道中
pub trait ISensorTunnelListener: Sync + Send {
    /// 调用openSensorAsControlling，成功或者失败会有回调
    fn on_open_failed(&self, tunnel: UniqueSensorTunnel, is_controlling: bool, err_reason: String);
    /// 代表有一个连接已经建立成功，is_controlling代表是发起连接的一端或者是被连接的一端
    /// 此时全双工的消息发送通道还未建立好，需要等onReadyToSend
    fn on_opened(&self, tunnel: UniqueSensorTunnel, is_controlling: bool);
    /// 代表了可以发送消息了,
    fn on_ready_to_send(&self, tunnel: UniqueSensorTunnel, ready_to_send: bool);
    /// 因为网络原因导致连接不稳定，内部正在进行重连，重连好了会调用onReadyToSend。
    fn on_open_tunnel_unstable(&self, tunnel: UniqueSensorTunnel, is_controlling: bool);
    /// 该状态代表即将发生端开的状态，该状态回调之后，接下去会回调onConnected或者onDisconnected（耗时长）。
    /// 建议收到该消息后，重新完成P2PRPC通道的重置，能够加快连接
    fn on_pre_closed(&self, tunnel: UniqueSensorTunnel, is_controlling: bool);
    /// 连接断开了
    fn on_closed(&self, tunnel: UniqueSensorTunnel, is_controlling: bool, err_reason: String);
    /// 收到了sensor通道的消息
    fn on_sensor_data_received(
        &self,
        tunnel: UniqueSensorTunnel,
        is_controlling: bool,
        sensor_msg_id: String,
        sensor_event: String,
        semsor_data: String,
    );
}

/// 每一对设备之间的连接都有唯一一个字节流数组通道，数据回调也在这个通道中
pub trait IByteBufListener: Sync + Send {
    /// 调用startByteBufChannel的时候，成功或者失败会有回调
    fn on_open_failed(&self, tunnel: UniqueByteBufTunnel, is_controlling: bool, err_reason: String);
    /// 代表有一个连接已经建立成功，is_controlling代表是发起连接的一端或者是被连接的一端
    /// 此时全双工的消息发送通道还未建立好，需要等onReadyToSend
    fn on_opened(&self, tunnel: UniqueByteBufTunnel, is_controlling: bool);
    /// 代表了可以发送消息了,
    fn on_ready_to_send(&self, tunnel: UniqueByteBufTunnel, ready_to_send: bool);
    /// 因为网络原因导致连接不稳定，内部正在进行重连，重连好了会调用onReadyToSend。
    fn on_open_tunnel_unstable(&self, tunnel: UniqueByteBufTunnel, is_controlling: bool);
    /// 该状态代表即将发生端开的状态，该状态回调之后，接下去会回调onConnected或者onDisconnected（耗时长）。
    /// 建议收到该消息后，重新完成P2PRPC通道的重置，能够加快连接
    fn on_pre_closed(&self, tunnel: UniqueByteBufTunnel, is_controlling: bool);
    /// 连接断开了
    fn on_closed(&self, tunnel: UniqueByteBufTunnel, is_controlling: bool, err_reason: String);
    /// 收到了bytebufer通道的消息
    fn on_bytebuf_received(
        &self,
        tunnel: UniqueByteBufTunnel,
        is_controlling: bool,
        byte_id: String,
        video: Vec<u8>,
        audio: Vec<u8>,
    );
}

/// 调用queryonline的回调
pub trait IQueryOnline: Sync + Send {
    /// 判断是否在线
    fn on_client_id_online(&self, success: bool, client_id: String, online: bool);
}

/// 酷连协议简单的API调用
pub trait ICoolinkEasyApi {
    /// 日志打印可以前置初始化
    fn pre_init_sdk_log(log_init: LogIntArgs);

    /// 初始化SDK
    fn init_easy_sdk(
        log_init: LogIntArgs,             // 日志初始化参数
        server_init: ServerQueryInitArgs, // 服务请求初始化参数
        httpd_init: HttpdInitArgs,        // httpd服务请求初始化参数
    );

    /// 设置消息通道的回调
    fn set_message_type_tunnel_listener(listener: Box<dyn IUniqueMessageTunnelListener>);

    /// 设置传感通道的回调
    fn set_sensor_type_tunnel_listener(listener: Box<dyn ISensorTunnelListener>);

    /// 设置字节数组通道的回调
    fn set_bytebuf_type_tunnel_listener(listener: Box<dyn IByteBufListener>);

    /// 重新设置用户，比如在切换账号之后
    fn re_init_user_id(user_id: String, // 新的用户id
    );

    /// 重新设置设备ID,比如在设备重新初始化之后
    fn re_init_device_id(device_id: String, // 新的设备id
    );

    /// 主动向对端设备发起消息通道的连接
    /// 返回int： 1 代表连接已经存在，默认是成功的，0代表此次成功，-1代表有错误发生
    fn connect_peer_as_controlling(remote_client_id: String, // 对端ID
    ) -> i32;

    /// 主动向对端设备发起sensor通道的连接
    /// 返回int： 1 代表连接已经存在，默认是成功的，0代表此次成功，-1代表有错误发生
    fn open_sensor_as_controlling(remote_client_id: String, // 对端ID
    ) -> i32;

    /// 主动向对端设备发起bytebuf通道的连接
    /// 返回int： 1 代表连接已经存在，默认是成功的，0代表此次成功，-1代表有错误发生
    fn start_byte_buf_channel_as_controlling(remote_client_id: String, // 对端ID
    ) -> i32;

    /// 判断消息通道是否已经准备完成
    fn is_message_tunnel_ready(remote_client_id: String, // 对端ID
    ) -> bool;

    /// 判断sensor通道是否已经准备完成
    fn is_sensor_tunnel_ready(remote_client_id: String, // 对端ID
    ) -> bool;

    /// 判断buffer通道是否已经准备完成
    fn is_bytebuf_tunnel_ready(remote_client_id: String, // 对端ID
    ) -> bool;

    /// 主动端开与对端的连接
    fn destroy_tunnel(remote_client_id: String, // 对端ID
    ) -> bool;

    /// 获得当前创建了P2P通道的连接设备列表
    fn get_connected_devices() -> Option<Vec<Option<UniqueTunnel>>>;

    /// 通过连接通道发送消息
    /// 该方法为异步方法，ISendMessageResult的message_id映射当前返回值
    /// 返回值代表了当前消息生成的message_id
    fn send_message(
        send_result: Box<dyn ISendMessageResult>, // 回调指针
        remote_client_id: String,                 // 对端设备ID
        message_event: String,                    // 消息名
        message: String,                          // 消息
    ) -> Result<String, anyhow::Error>; // 成功或者错误，成功返回message_id

    /// 通过连接通道发送广播消息，不需要带上remote_client_id参数
    /// 该方法为异步方法，ISendMessageResult的message_id映射当前返回值
    /// 返回值代表了一组(remote_client_id,message_id)的结果
    fn broadcast_message(
        send_result: Box<dyn ISendMessageResult>, //结果回调
        message_event: String,                    //消息名称
        message: String,                          // 消息内容
    ) -> Vec<Result<(String, String), anyhow::Error>>; // 成功或者错误，成功返回(remote_client_id,message_id)消息列表

    /// 通过sensor通道发送消息
    /// 该方法为异步方法，ISendSensorResult的message_id映射当前返回值
    /// 返回值代表了当前消息生成的message_id
    fn send_sensor(
        send_result: Box<dyn ISendSensorResult>, //回调结果
        remote_client_id: String,                // 对端设备ID
        sensor_event: String,                    // sensor消息名称
        sensor_data: String,                     // sensor消息
    ) -> Result<String, anyhow::Error>; // 成功或者错误，成功返回message_id

    /// 通过流数组通道发送消息
    /// 该方法为异步方法，ISendSensorResult的message_id映射当前返回值
    /// 返回值代表了当前消息生成的message_id
    fn send_bytebuf(
        send_result: Box<dyn ISendByteBufResult>, //结果回调
        remote_client_id: String,                 // 对端设备ID
        can_lost: bool,                           // 能否丢包
        video_buf: Option<Vec<u8>>,               // 视频数据
        audio_buf: Option<Vec<u8>>,               // 音频数据
    ) -> Result<String, anyhow::Error>; // 成功或者错误，成功返回message_id

    /// 将本地文件添加到P2P的mapping中
    fn generate_p2prpc_file_mapping(
        file_path: String, // 本地文件的绝对路径
    ) -> Result<String, anyhow::Error>; // 成功返回mapping ID

    /// 获取到对端发来的mapping文件，生成url
    fn get_p2prpc_file_mapping_to_http(
        remote_client_id: String, // 对方的设备id
        mapping_url: String,      // 映射的url名字
    ) -> Result<String, anyhow::Error>; // 成功返回和http ID

    /// 从网关处请求判断client_id是否是在线状态
    fn query_online_from_gateway(
        client_id: String,                     // 接入的client_id
        query_callback: Box<dyn IQueryOnline>, // 接口请求回调
    ) -> Result<(), anyhow::Error>;

    /// 判断是否在同一个host下
    fn is_in_same_host(remote_client_id: String, //远端client_id
    ) -> bool;
}

struct EasyP2PRPCServiceCallback {}
impl P2PRPCServiceCallback for EasyP2PRPCServiceCallback {
    /// received p2prcp server created event
    fn on_p2prpc_server_created(
        &self,
        local_client_id: String, //always equal to on_sg_online return local_client_id
    ) -> i64 {
        log::info!(
            "on_p2prpc_server_created,local_client_id = {}",
            local_client_id
        );
        set_p2pprc_started(true);
        notify_p2prpc_srv_started(local_client_id.to_owned());

        0
    }

    /// received p2prcp server droped event
    fn on_p2prpc_server_droped(
        &self,
        reason: String, // p2prpc server dropped with some reason
    ) -> i64 {
        log::info!("on_p2prpc_server_droped,reason = {}", reason);
        set_p2pprc_started(false);

        clear_unique_msg_tunnel_map();
        clear_unique_sensor_tunnel_map();
        clear_unique_bytebuf_tunnel_map();
        clear_p2p_rtc_pair_map();
        clear_control_kind_map();

        0
    }

    /// received p2prpc data transport tunnel created, a p2prpc server can handle many tunnels.
    fn on_p2prpc_tunnel_created(
        &self,
        tunnel_id: String,        //  p2prpc tunnel of unique tunnel id
        local_client_id: String,  // local_client_id for p2prpc tunnel
        remote_client_id: String, // remote_client_id for p2prpc tunnel
        is_controlling: bool,     // check if local_client_id is controlling role.
    ) -> i64 {
        log::info!(
            "on_p2prpc_tunnel_created tunnel_id ={},local_client_id ={},remote_client_id={},is_controlling={}",
            tunnel_id,
            local_client_id,
            remote_client_id,
            is_controlling,
        );
        let mut control_type = if is_controlling {
            ControlKind::CONTROLLING
        } else {
            ControlKind::CONTROLLED
        };
        get_or_set_control_kind_map(true, tunnel_id.clone(), Some(control_type.clone()));
        if !is_pairmap_contains(tunnel_id.clone()) {
            let tunnel_data_type = tunnel_data_type_kind(tunnel_id.clone());
            get_or_set_p2p_rtc_pair_map(
                true,
                tunnel_id.clone(),
                Some((
                    P2PRPCTunnelPair::new(
                        tunnel_id.clone(),
                        local_client_id,
                        remote_client_id,
                        control_type,
                        tunnel_data_type,
                    ),
                    TunnelConnectionState::New,
                )),
            );
        }
        let (pair, state) = get_or_set_p2p_rtc_pair_map(false, tunnel_id.clone(), None)
            .unwrap()
            .unwrap();
        if pair.control_type == ControlKind::CONTROLLING {
            if pair.tunnel_data_type == TunnelDataTypeKind::LINKED
                || pair.tunnel_data_type == TunnelDataTypeKind::SENSOR
            {
                log::info!("on_p2prpc_tunnel_created select message tunnel as duplex.");
                // 此处是的控制端直接就调用select_text_tunnel_duplex,实际上可能没有什么用处
                // 需要等到connected状态回调的时候才有用
                P2PRPCServiceContractImp::select_text_tunnel_duplex(tunnel_id.clone());
            } else if pair.tunnel_data_type == TunnelDataTypeKind::BYTEBUF {
                log::info!("on_p2prpc_tunnel_created select bytebuf tunnel as duplex.");
                // 此处是的控制端直接就调用select_bytebuf_tunnel_duplex,实际上可能没有什么用处
                // 需要等到connected状态回调的时候才有用
                P2PRPCServiceContractImp::select_bytebuf_tunnel_duplex(tunnel_id.clone());
            } else {
                log::info!(
                    "on_p2prpc_tunnel_created and data_type = {:?}.",
                    pair.tunnel_data_type
                );
            }
        } else {
            log::info!("on_p2prpc_tunnel_created ,but is controlled type,wait for controlling select duplex.");
        }

        if pair.tunnel_data_type == TunnelDataTypeKind::LINKED {
            if let Ok(unique_msg_tunnel_opt) =
                find_unique_msg_tunnel_by_tunnel_id(tunnel_id.clone())
            {
                log::info!("on_p2prpc_tunnel_created find_unique_msg_tunnel_by_tunnel_id success.");
                if let Some(unique_msg_tunnel) = unique_msg_tunnel_opt {
                    // if find messge tunnel already!
                    let ((old_pair, state, listener)) =
                        get_or_set_unique_msg_tunnel_map(false, unique_msg_tunnel.clone(), None)
                            .unwrap()
                            .unwrap();
                    listener.on_connected(unique_msg_tunnel.clone(), old_pair.is_controlling());

                    let mut new_pair = old_pair.clone();
                    new_pair.is_first_time_inited = false;
                    get_or_set_unique_msg_tunnel_map(
                        true,
                        unique_msg_tunnel.to_owned(),
                        Some((new_pair, state, listener)),
                    );
                } else {
                    // is controlled type
                    log::warn!("on_p2prpc_tunnel_created find_unique_msg_tunnel_by_tunnel_id false,iscontrolled = {:?} wait for select duplex.",pair.is_controlling());
                }
            };
            log::info!("on_p2prpc_tunnel_created find_unique_msg_tunnel_by_tunnel_id end.")
        } else if pair.tunnel_data_type == TunnelDataTypeKind::SENSOR {
            if let Ok(unique_sensor_tunnel_opt) =
                find_unique_sensor_tunnel_by_tunnel_id(tunnel_id.clone())
            {
                log::info!(
                    "on_p2prpc_tunnel_created find_unique_sensor_tunnel_by_tunnel_id success."
                );
                if let Some(unique_sensor_tunnel) = unique_sensor_tunnel_opt {
                    // if find messge tunnel already!
                    let ((old_pair, state, listener)) = get_or_set_unique_sensor_tunnel_map(
                        false,
                        unique_sensor_tunnel.clone(),
                        None,
                    )
                    .unwrap()
                    .unwrap();
                    listener.on_opened(unique_sensor_tunnel.clone(), old_pair.is_controlling());

                    let mut new_pair = old_pair.clone();
                    new_pair.is_first_time_inited = false;
                    get_or_set_unique_sensor_tunnel_map(
                        true,
                        unique_sensor_tunnel.to_owned(),
                        Some((new_pair, state, listener)),
                    );
                } else {
                    // is controlled type
                    log::warn!("on_p2prpc_tunnel_created find_unique_sensor_tunnel_by_tunnel_id false,controlling type = {:?}.",pair.is_controlling());
                }
            };
            log::info!("on_p2prpc_tunnel_created find_unique_sensor_tunnel_by_tunnel_id end.")
        } else if pair.tunnel_data_type == TunnelDataTypeKind::BYTEBUF {
            if let Ok(unique_bytebuf_tunnel_opt) =
                find_unique_bytebuf_tunnel_by_tunnel_id(tunnel_id.clone())
            {
                log::info!(
                    "on_p2prpc_tunnel_created find_unique_bytebuf_tunnel_by_tunnel_id success."
                );
                if let Some(unique_bytebuf_tunnel) = unique_bytebuf_tunnel_opt {
                    // if find messge tunnel already!
                    let ((old_pair, state, listener)) = get_or_set_unique_bytebuf_tunnel_map(
                        false,
                        unique_bytebuf_tunnel.clone(),
                        None,
                    )
                    .unwrap()
                    .unwrap();
                    listener.on_opened(unique_bytebuf_tunnel.clone(), old_pair.is_controlling());

                    let mut new_pair = old_pair.clone();
                    new_pair.is_first_time_inited = false;
                    get_or_set_unique_bytebuf_tunnel_map(
                        true,
                        unique_bytebuf_tunnel.to_owned(),
                        Some((new_pair, state, listener)),
                    );
                } else {
                    // is controlled type
                    log::warn!("on_p2prpc_tunnel_created find_unique_bytebuf_tunnel_by_tunnel_id false,controlling type = {:?}.",pair.is_controlling());
                }
            };
            log::info!("on_p2prpc_tunnel_created find_unique_bytebuf_tunnel_by_tunnel_id end.")
        }

        0
    }
    /// received p2prpc pair tunnel destroyed
    fn on_p2prpc_tunnel_destroy(
        &self,
        tunnel_id: String,        //  p2prpc tunnel of unique tunnel id
        local_client_id: String,  // local_client_id for p2prpc tunnel
        remote_client_id: String, // remote_client_id for p2prpc tunnel
    ) -> i64 {
        log::info!(
            "on_p2prpc_tunnel_destroy tunnel_id ={},local_client_id ={},remote_client_id={}",
            tunnel_id,
            local_client_id,
            remote_client_id
        );
        get_or_set_control_kind_map(true, tunnel_id.clone(), None);
        get_or_set_p2p_rtc_pair_map(true, tunnel_id.clone(), None);
        delete_unique_msg_tunnel_map(tunnel_id.clone());
        delete_unique_sensor_tunnel_map(tunnel_id.clone());
        delete_unique_bytebuf_tunnel_map(tunnel_id.clone());

        0
    }
    /// received ice-kcp tunnel state changged
    fn on_p2prpc_tunnel_state_changed(
        &self,
        tunnel_id: String,        //  p2prpc tunnel of unique tunnel id
        local_client_id: String,  // local_client_id for p2prpc tunnel
        remote_client_id: String, // remote_client_id for p2prpc tunnel
        state: String,            // p2prpc tunnel current state
        state_msg: String,
    ) -> i64 {
        refresh_pair_map_state(
            tunnel_id.clone(),
            local_client_id.clone(),
            remote_client_id.clone(),
            state.clone(),
            state_msg.clone(),
        );
        refresh_unique_msg_map_state(
            tunnel_id.clone(),
            local_client_id.clone(),
            remote_client_id.clone(),
            state.clone(),
            state_msg.clone(),
        );
        refresh_unique_sensor_map_state(
            tunnel_id.clone(),
            local_client_id.clone(),
            remote_client_id.clone(),
            state.clone(),
            state_msg.clone(),
        );
        refresh_unique_bytebuf_map_state(
            tunnel_id.clone(),
            local_client_id.clone(),
            remote_client_id.clone(),
            state.clone(),
            state_msg.clone(),
        );

        0
    }
    /// received ice-kcp tunnel pair changged
    fn on_p2prpc_tunnel_pair_changed(
        &self,
        tunnel_id: String,        //  p2prpc tunnel of unique tunnel id
        local_client_id: String,  // local_client_id for p2prpc tunnel
        remote_client_id: String, // remote_client_id for p2prpc tunnel
        mine_pair: String,        // my tunnel ip/port pair
        remote_pair: String,      // remote tunnel ip/port pair
    ) -> i64 {
        log::info!(
            "on_p2prpc_tunnel_pair_changed tunnel_id = {:?}, mine_pair = {:?},remote_pair = {:?}",
            tunnel_id.clone(),
            mine_pair.clone(),
            remote_pair
        );

        refresh_pair_map_state(
            tunnel_id.clone(),
            local_client_id.clone(),
            remote_client_id.clone(),
            "Completed".to_owned(),
            "".to_owned(),
        );
        refresh_unique_msg_map_state(
            tunnel_id.clone(),
            local_client_id.clone(),
            remote_client_id.clone(),
            "Completed".to_owned(),
            "".to_owned(),
        );
        refresh_unique_sensor_map_state(
            tunnel_id.clone(),
            local_client_id.clone(),
            remote_client_id.clone(),
            "Completed".to_owned(),
            "".to_owned(),
        );
        refresh_unique_bytebuf_map_state(
            tunnel_id.clone(),
            local_client_id.clone(),
            remote_client_id.clone(),
            "Completed".to_owned(),
            "".to_owned(),
        );

        0
    }

    /// received text tunnel selected as duplex rpc tunnel
    fn on_text_tunnel_select_duplex(
        &self,
        tunnel_id: String,        // p2prpc tunnel of unique tunnel id
        local_client_id: String,  //  local_client_id for p2prpc tunnel
        remote_device_id: String, // remote_client_id for p2prpc tunnel
    ) -> i64 {
        log::info!(
            "on_text_tunnel_select_duplex tunnel_id ={},local_client_id ={},remote_device_id={}",
            tunnel_id,
            local_client_id,
            remote_device_id
        );
        if let Ok(pair_result) = get_or_set_p2p_rtc_pair_map(false, tunnel_id.clone(), None) {
            if let Some((pair, state)) = pair_result {
                if pair.tunnel_data_type == TunnelDataTypeKind::LINKED {
                    if let Ok(remote_tunnels_vec) =
                        find_unique_msg_tunnel_by_remote_c_id(remote_device_id.clone())
                    {
                        for (_, u_tunnel_opt) in remote_tunnels_vec.iter().enumerate() {
                            if let Some(u_tunnel) = u_tunnel_opt {
                                let remote_unique_tunnel_id = u_tunnel.tunnel_id.clone(); //         if !remote_unique_tunnel_id.eq_ignore_ascii_case(&tunnel_id) {
                                                                                          // 如果当前找到连接的对端有多条的连接通道，那么代表对端已经发生了重连，需要将之前管理的连接提前断开
                                if !remote_unique_tunnel_id.eq_ignore_ascii_case(&tunnel_id) {
                                    log::warn!("on_text_tunnel_select_duplex received and tunnel_id not equal, destroyed :{:?}",remote_unique_tunnel_id.clone());
                                    P2PRPCServiceContractImp::drop_p2prpc_tunnel(
                                        remote_unique_tunnel_id.to_owned(),
                                    );
                                    get_or_set_unique_msg_tunnel_map(true, u_tunnel.clone(), None);
                                }
                            }
                        }
                    }

                    if let Ok(unique_msg_tunnel_opt) =
                        find_unique_msg_tunnel_by_tunnel_id(tunnel_id.clone())
                    {
                        log::info!("on_text_tunnel_select_duplex find_unique_msg_tunnel_by_tunnel_id success.");
                        if let Some(unique_msg_tunnel) = unique_msg_tunnel_opt {
                            // if find messge tunnel already!
                            let ((old_pair, state, listener)) = get_or_set_unique_msg_tunnel_map(
                                false,
                                unique_msg_tunnel.clone(),
                                None,
                            )
                            .unwrap()
                            .unwrap();
                            if old_pair.is_first_time_inited {
                                listener.on_ready_to_send(
                                    unique_msg_tunnel.clone(),
                                    old_pair.is_controlling(),
                                );
                            } else {
                                listener.on_connected(
                                    unique_msg_tunnel.clone(),
                                    old_pair.is_controlling(),
                                );
                            }

                            let mut new_pair = old_pair.clone();
                            new_pair.is_first_time_inited = false;
                            get_or_set_unique_msg_tunnel_map(
                                true,
                                unique_msg_tunnel.to_owned(),
                                Some((new_pair, state, listener)),
                            );
                        } else {
                            // is controlled type
                            log::warn!("on_text_tunnel_select_duplex find_unique_msg_tunnel_by_tunnel_id false,iscontrolled = {:?} wait for select duplex.",pair.is_controlling());
                            let new_unique_text_tunnel = UniqueTunnel {
                                tunnel_id,
                                remote_id: remote_device_id,
                            };
                            match UNIQUE_MSG_TUNNEL_CB.try_get() {
                                Some(listener_instance) => {
                                    let listener = listener_instance.read().unwrap().clone();
                                    let state = TunnelConnectionState::Completed;
                                    get_or_set_unique_msg_tunnel_map(
                                        true,
                                        new_unique_text_tunnel.clone(),
                                        Some((pair.clone(), state, listener)),
                                    );
                                }
                                None => {
                                    log::error!("on_text_tunnel_select_duplex error! please call set_message_type_tunnel_listener first!");
                                }
                            };
                        }
                    };
                    log::info!(
                        "on_text_tunnel_select_duplex find_unique_msg_tunnel_by_tunnel_id end."
                    )
                } else if pair.tunnel_data_type == TunnelDataTypeKind::SENSOR {
                    if let Ok(remote_tunnels_vec) =
                        find_unique_sensor_tunnel_by_remote_c_id(remote_device_id.clone())
                    {
                        for (_, u_tunnel_opt) in remote_tunnels_vec.iter().enumerate() {
                            if let Some(u_tunnel) = u_tunnel_opt {
                                let remote_unique_tunnel_id = u_tunnel.tunnel_id.clone();
                                if !remote_unique_tunnel_id.eq_ignore_ascii_case(&tunnel_id) {
                                    // 如果当前找到连接的对端有多条的连接通道，那么代表对端已经发生了重连，需要将之前管理的连接提前断开
                                    log::warn!("on_text_tunnel_select_duplex sensor received and tunnel_id not equal, destroyed :{:?}",remote_unique_tunnel_id.clone());
                                    P2PRPCServiceContractImp::drop_p2prpc_tunnel(
                                        remote_unique_tunnel_id.to_owned(),
                                    );
                                    get_or_set_unique_sensor_tunnel_map(
                                        true,
                                        u_tunnel.clone(),
                                        None,
                                    );
                                }
                            }
                        }
                    }
                    if let Ok(unique_sensor_tunnel_opt) =
                        find_unique_sensor_tunnel_by_tunnel_id(tunnel_id.clone())
                    {
                        log::info!("on_text_tunnel_select_duplex find_unique_sensor_tunnel_by_tunnel_id success.");
                        if let Some(unique_sensor_tunnel) = unique_sensor_tunnel_opt {
                            // if find sensor tunnel already!
                            let ((old_pair, state, listener)) =
                                get_or_set_unique_sensor_tunnel_map(
                                    false,
                                    unique_sensor_tunnel.clone(),
                                    None,
                                )
                                .unwrap()
                                .unwrap();
                            if old_pair.is_first_time_inited {
                                listener.on_ready_to_send(
                                    unique_sensor_tunnel.clone(),
                                    old_pair.is_controlling(),
                                );
                            } else {
                                listener.on_opened(
                                    unique_sensor_tunnel.clone(),
                                    old_pair.is_controlling(),
                                );
                            }
                            let mut new_pair = old_pair.clone();
                            new_pair.is_first_time_inited = false;
                            get_or_set_unique_sensor_tunnel_map(
                                true,
                                unique_sensor_tunnel.to_owned(),
                                Some((new_pair, state, listener)),
                            );
                        } else {
                            // is controlled type
                            log::warn!("on_text_tunnel_select_duplex find_unique_sensor_tunnel_by_tunnel_id false,iscontrolled = {:?} wait for select duplex.",pair.is_controlling());
                            let new_unique_sensor_tunnel = UniqueSensorTunnel {
                                tunnel_id,
                                remote_id: remote_device_id,
                            };
                            match UNIQUE_SENSOR_TUNNEL_CB.try_get() {
                                Some(listener_instance) => {
                                    let listener = listener_instance.read().unwrap().clone();
                                    let state = TunnelConnectionState::Completed;
                                    get_or_set_unique_sensor_tunnel_map(
                                        true,
                                        new_unique_sensor_tunnel.to_owned(),
                                        Some((pair.clone(), state, listener)),
                                    );
                                }
                                None => {
                                    log::error!("on_text_tunnel_select_duplex sensor error! please call set_bytebuf_type_tunnel_listener first!");
                                }
                            };
                        }
                    };
                    log::info!(
                        "on_text_tunnel_select_duplex find_unique_sensor_tunnel_by_tunnel_id end."
                    )
                } else {
                    log::info!("on_text_tunnel_select_duplex unknow tunnel_data_type !")
                }
            }
        }

        0
    }

    /// received text tunnel canceled as duplex rpc tunnel
    fn on_text_tunnel_cancel_duplex(
        &self,
        tunnel_id: String, // p2prpc tunnel of unique tunnel id
    ) -> i64 {
        log::info!("on_text_tunnel_cancel_duplex tunnel_id ={}", tunnel_id);
        if let Ok(pair_result) = get_or_set_p2p_rtc_pair_map(false, tunnel_id.clone(), None) {
            if let Some((pair, state)) = pair_result {
                if pair.tunnel_data_type == TunnelDataTypeKind::LINKED {
                    log::info!("on_text_tunnel_cancel_duplex select message tunnel as duplex.");
                    P2PRPCServiceContractImp::select_text_tunnel_duplex(tunnel_id.clone());
                } else if pair.tunnel_data_type == TunnelDataTypeKind::SENSOR {
                    P2PRPCServiceContractImp::select_text_tunnel_duplex(tunnel_id.clone());
                    log::info!("on_text_tunnel_cancel_duplex select sensor tunnel as duplex.")
                } else {
                    log::info!("on_text_tunnel_cancel_duplex unknow tunnel_data_type !")
                }
            }
        }

        0
    }

    /// received text message from tunnel
    fn on_p2prpc_tunnel_rec_msg(
        &self,
        tunnel_id: String,        // p2prpc tunnel of unique tunnel id
        local_client_id: String,  //  local_client_id of this p2prpc pair tunnel
        remote_client_id: String, // remote_device_id of this p2prpc pair tunnel
        msg_id: String,           // unique msgid received
        event: String,            // message event received
        data: String,             // message data received
    ) -> i64 {
        log::info!("on_p2prpc_tunnel_rec_msg tunnel_id ={},local_client_id ={},remote_client_id={},msg_id={},event={},data={}", tunnel_id, local_client_id,remote_client_id,msg_id, event,data);
        if let Ok(pair_result) = get_or_set_p2p_rtc_pair_map(false, tunnel_id.clone(), None) {
            if let Some((pair, state)) = pair_result {
                if pair.tunnel_data_type == TunnelDataTypeKind::LINKED {
                    if let Ok(unique_msg_tunnel_opt) =
                        find_unique_msg_tunnel_by_tunnel_id(tunnel_id.clone())
                    {
                        if let Some(unique_msg_tunnel) = unique_msg_tunnel_opt {
                            let ((pair, state, listener)) = get_or_set_unique_msg_tunnel_map(
                                false,
                                unique_msg_tunnel.clone(),
                                None,
                            )
                            .unwrap()
                            .unwrap();
                            listener.on_received(
                                unique_msg_tunnel.clone(),
                                pair.is_controlling(),
                                msg_id,
                                event,
                                data,
                            );
                        } else {
                            log::info!("on_p2prpc_tunnel_rec_msg find_unique_msg_tunnel_by_tunnel_id null.");
                            if let Ok(remote_tunnels_vec) =
                                find_unique_msg_tunnel_by_remote_c_id(remote_client_id.clone())
                            {
                                for (_, u_tunnel_opt) in remote_tunnels_vec.iter().enumerate() {
                                    if let Some(u_tunnel) = u_tunnel_opt {
                                        let remote_unique_tunnel_id = u_tunnel.tunnel_id.clone();
                                        if !remote_unique_tunnel_id.eq_ignore_ascii_case(&tunnel_id)
                                        {
                                            // 如果当前找到连接的对端有多条的连接通道，那么代表对端已经发生了重连，需要将之前管理的连接提前断开
                                            log::warn!("on_p2prpc_tunnel_rec_msg received and tunnel_id not equal, destroyed :{:?}",remote_unique_tunnel_id.clone());
                                            P2PRPCServiceContractImp::drop_p2prpc_tunnel(
                                                remote_unique_tunnel_id.to_owned(),
                                            );
                                            get_or_set_unique_msg_tunnel_map(
                                                true,
                                                u_tunnel.clone(),
                                                None,
                                            );
                                        }
                                    }
                                }
                            }
                            log::warn!("on_p2prpc_tunnel_rec_msg received and tunnel_id not equal, destroyed :{:?}",tunnel_id.clone());
                            P2PRPCServiceContractImp::drop_p2prpc_tunnel(tunnel_id.clone());
                        }
                    } else {
                        log::error!("on_p2prpc_tunnel_rec_msg find_unique_msg_tunnel_by_tunnel_id unwrap err!");
                    }
                } else if pair.tunnel_data_type == TunnelDataTypeKind::SENSOR {
                    if let Ok(unique_sensor_tunnel_opt) =
                        find_unique_sensor_tunnel_by_tunnel_id(tunnel_id.clone())
                    {
                        if let Some(unique_sensor_tunnel) = unique_sensor_tunnel_opt {
                            let ((pair, state, listener)) = get_or_set_unique_sensor_tunnel_map(
                                false,
                                unique_sensor_tunnel.clone(),
                                None,
                            )
                            .unwrap()
                            .unwrap();
                            listener.on_sensor_data_received(
                                unique_sensor_tunnel.clone(),
                                pair.is_controlling(),
                                msg_id,
                                event,
                                data,
                            );
                        } else {
                            log::info!("on_p2prpc_tunnel_rec_msg find_unique_sensor_tunnel_by_tunnel_id null.");
                            if let Ok(remote_tunnels_vec) =
                                find_unique_sensor_tunnel_by_remote_c_id(remote_client_id.clone())
                            {
                                for (_, u_tunnel_opt) in remote_tunnels_vec.iter().enumerate() {
                                    if let Some(u_tunnel) = u_tunnel_opt {
                                        let remote_unique_tunnel_id = u_tunnel.tunnel_id.clone();
                                        if !remote_unique_tunnel_id.eq_ignore_ascii_case(&tunnel_id)
                                        {
                                            // 如果当前找到连接的对端有多条的连接通道，那么代表对端已经发生了重连，需要将之前管理的连接提前断开
                                            log::warn!("on_p2prpc_tunnel_rec_msg received and senor tunnel_id not equal, destroyed :{:?}",remote_unique_tunnel_id.clone());
                                            P2PRPCServiceContractImp::drop_p2prpc_tunnel(
                                                remote_unique_tunnel_id.to_owned(),
                                            );
                                            get_or_set_unique_sensor_tunnel_map(
                                                true,
                                                u_tunnel.clone(),
                                                None,
                                            );
                                        }
                                    }
                                }
                            }
                            log::warn!("on_p2prpc_tunnel_rec_msg received and senor tunnel_id not equal, destroyed :{:?}",tunnel_id.clone());
                            P2PRPCServiceContractImp::drop_p2prpc_tunnel(tunnel_id.clone());
                        }
                    } else {
                        log::error!("on_p2prpc_tunnel_rec_msg find_unique_sensor_tunnel_by_tunnel_id unwrap err!");
                    }
                } else {
                    log::error!(
                        "on_p2prpc_tunnel_rec_msg channel type error: {:?} ,destroyed.",
                        pair.tunnel_data_type
                    );
                    P2PRPCServiceContractImp::drop_p2prpc_tunnel(tunnel_id.clone());
                }
            }
        }

        0
    }

    /// received byte buffer message from tunnel
    fn on_p2prpc_tunnel_rec_bytebuffer(
        &self,
        tunnel_id: String,  // p2prpc tunnel of unique tunnel id
        buffer_id: String,  // unique bytebuffer id received
        video_buf: Vec<u8>, // video bytebuffer received
        audio_buf: Vec<u8>, // audio bytebuffer received
    ) -> i64 {
        log::info!("on_p2prpc_tunnel_rec_bytebuffer ");
        if let Ok(pair_result) = get_or_set_p2p_rtc_pair_map(false, tunnel_id.clone(), None) {
            if let Some((pair, state)) = pair_result {
                if pair.tunnel_data_type == TunnelDataTypeKind::BYTEBUF {
                    if let Ok(unique_bytebuf_tunnel_opt) =
                        find_unique_bytebuf_tunnel_by_tunnel_id(tunnel_id.clone())
                    {
                        if let Some(unique_bytebuf_tunnel) = unique_bytebuf_tunnel_opt {
                            let ((pair, state, listener)) = get_or_set_unique_bytebuf_tunnel_map(
                                false,
                                unique_bytebuf_tunnel.clone(),
                                None,
                            )
                            .unwrap()
                            .unwrap();
                            listener.on_bytebuf_received(
                                unique_bytebuf_tunnel.clone(),
                                pair.is_controlling(),
                                buffer_id,
                                video_buf,
                                audio_buf,
                            );
                        } else {
                            log::info!("on_p2prpc_tunnel_rec_bytebuffer find_unique_bytebuf_tunnel_by_tunnel_id null.");
                            let remote_client_id = pair.remote_client_id.clone();
                            if let Ok(remote_tunnels_vec) =
                                find_unique_bytebuf_tunnel_by_remote_c_id(remote_client_id.clone())
                            {
                                for (_, u_tunnel_opt) in remote_tunnels_vec.iter().enumerate() {
                                    if let Some(u_tunnel) = u_tunnel_opt {
                                        let remote_unique_tunnel_id = u_tunnel.tunnel_id.clone();
                                        if !remote_unique_tunnel_id.eq_ignore_ascii_case(&tunnel_id)
                                        {
                                            // 如果当前找到连接的对端有多条的连接通道，那么代表对端已经发生了重连，需要将之前管理的连接提前断开
                                            log::warn!("on_p2prpc_tunnel_rec_bytebuffer received and tunnel_id not equal, destroyed :{:?}",remote_unique_tunnel_id.clone());
                                            P2PRPCServiceContractImp::drop_p2prpc_tunnel(
                                                remote_unique_tunnel_id.to_owned(),
                                            );
                                            get_or_set_unique_bytebuf_tunnel_map(
                                                true,
                                                u_tunnel.clone(),
                                                None,
                                            );
                                        }
                                    }
                                }
                            }
                            log::warn!("on_p2prpc_tunnel_rec_bytebuffer received and tunnel_id not equal, destroyed :{:?}",tunnel_id.clone());
                            P2PRPCServiceContractImp::drop_p2prpc_tunnel(tunnel_id.clone());
                        }
                    } else {
                        log::error!("on_p2prpc_tunnel_rec_bytebuffer find_unique_bytebuf_tunnel_by_tunnel_id unwrap err!");
                    }
                } else {
                    log::error!(
                        "on_p2prpc_tunnel_rec_bytebuffer channel type error: {:?} ,destroyed.",
                        pair.tunnel_data_type
                    );
                    P2PRPCServiceContractImp::drop_p2prpc_tunnel(tunnel_id.clone());
                }
            }
        }

        0
    }

    /// send message with text tunnel replay result
    fn on_p2prpc_send_msg_result(
        &self,
        msg_id: String,    // unique msgid sent
        tunnel_id: String, // p2prpc tunnel of unique tunnel id
        is_ok: bool,       // message sent is ok
        err_msg: String,   // if is_ok == false,error message print
    ) -> i64 {
        if let Ok(pair_result) = get_or_set_p2p_rtc_pair_map(false, tunnel_id.clone(), None) {
            if let Some((pair, state)) = pair_result {
                if pair.tunnel_data_type == TunnelDataTypeKind::LINKED {
                    log::info!(
                        "on_p2prpc_send_msg_result linked msg_id ={},tunnel_id ={},is_ok={},msg={}",
                        msg_id,
                        tunnel_id,
                        is_ok,
                        err_msg
                    );
                    callback_msg_tunnel_message(msg_id, is_ok, err_msg);
                } else if pair.tunnel_data_type == TunnelDataTypeKind::SENSOR {
                    log::info!(
                        "on_p2prpc_send_msg_result sensor msg_id ={},tunnel_id ={},is_ok={},msg={}",
                        msg_id,
                        tunnel_id,
                        is_ok,
                        err_msg
                    );
                    callback_sensor_tunnel_message(msg_id, is_ok, err_msg);
                } else {
                    log::info!("on_p2prpc_send_msg_result error tunnel type!");
                }
            }
        }

        0
    }

    /// send bytebuffer with bytebuffer tunnel replay result
    fn on_p2prpc_send_bytebuffer_result(
        &self,
        buffer_id: String, // unique bytebuffer sent
        tunnel_id: String, // p2prpc tunnel of unique tunnel id
        is_ok: bool,       // bytebuffer sent is ok
        err_msg: String,   // if is_ok == false,error message print
    ) -> i64 {
        if let Ok(pair_result) = get_or_set_p2p_rtc_pair_map(false, tunnel_id.clone(), None) {
            if let Some((pair, state)) = pair_result {
                if pair.tunnel_data_type == TunnelDataTypeKind::BYTEBUF {
                    log::info!(
                        "on_p2prpc_send_bytebuffer_result buffer_id ={},tunnel_id ={},is_ok={},msg={}",
                        buffer_id,
                        tunnel_id,
                        is_ok,
                        err_msg
                    );
                    callback_bytebuf_tunnel_message(buffer_id, is_ok, err_msg);
                } else {
                    log::info!("on_p2prpc_send_bytebuffer_result error tunnel type!");
                }
            }
        }
        0
    }

    /// received bytebuffer tunnel selected as duplex rpc tunnel
    fn on_bybebuf_tunnel_select_duplex(
        &self,
        tunnel_id: String,        // p2prpc tunnel of unique tunnel id
        local_client_id: String,  //  local_client_id of this p2prpc pair tunnel
        remote_device_id: String, //  remote_device_id of this p2prpc pair tunnel
    ) -> i64 {
        log::info!("on_bybebuf_tunnel_select_duplex ");
        if let Ok(pair_result) = get_or_set_p2p_rtc_pair_map(false, tunnel_id.clone(), None) {
            if let Some((pair, state)) = pair_result {
                if pair.tunnel_data_type == TunnelDataTypeKind::BYTEBUF {
                    if let Ok(remote_tunnels_vec) =
                        find_unique_bytebuf_tunnel_by_remote_c_id(remote_device_id.clone())
                    {
                        for (_, u_tunnel_opt) in remote_tunnels_vec.iter().enumerate() {
                            if let Some(u_tunnel) = u_tunnel_opt {
                                if !u_tunnel.tunnel_id.eq_ignore_ascii_case(&tunnel_id) {
                                    // 如果当前找到连接的对端有多条的连接通道，那么代表对端已经发生了重连，需要将之前管理的连接提前断开
                                    log::warn!("on_bybebuf_tunnel_select_duplex bytebuf received and tunnel_id not equal, destroyed :{:?}",tunnel_id);
                                    P2PRPCServiceContractImp::drop_p2prpc_tunnel(tunnel_id.clone());
                                    get_or_set_unique_bytebuf_tunnel_map(
                                        true,
                                        u_tunnel.clone(),
                                        None,
                                    );
                                }
                            }
                        }
                    }
                    if let Ok(unique_bytebuf_tunnel_opt) =
                        find_unique_bytebuf_tunnel_by_tunnel_id(tunnel_id.clone())
                    {
                        log::info!("on_bybebuf_tunnel_select_duplex find_unique_bytebuf_tunnel_by_tunnel_id success.");
                        if let Some(unique_bytebuf_tunnel) = unique_bytebuf_tunnel_opt {
                            // if find bytebuf tunnel already!
                            let ((old_pair, state, listener)) =
                                get_or_set_unique_bytebuf_tunnel_map(
                                    false,
                                    unique_bytebuf_tunnel.clone(),
                                    None,
                                )
                                .unwrap()
                                .unwrap();
                            if old_pair.is_first_time_inited {
                                listener.on_ready_to_send(
                                    unique_bytebuf_tunnel.clone(),
                                    old_pair.is_controlling(),
                                );
                            } else {
                                listener.on_opened(
                                    unique_bytebuf_tunnel.clone(),
                                    old_pair.is_controlling(),
                                );
                            }
                            let mut new_pair = old_pair.clone();
                            new_pair.is_first_time_inited = false;
                            get_or_set_unique_bytebuf_tunnel_map(
                                true,
                                unique_bytebuf_tunnel.to_owned(),
                                Some((new_pair, state, listener)),
                            );
                        } else {
                            // is controlled type
                            log::warn!("on_bybebuf_tunnel_select_duplex find_unique_bytebuf_tunnel_by_tunnel_id false,iscontrolled = {:?} wait for select duplex.",pair.is_controlling());
                            let new_unique_bytebuf_tunnel = UniqueByteBufTunnel {
                                tunnel_id,
                                remote_id: remote_device_id,
                            };
                            match UNIQUE_BYTEBUF_TUNNEL_CB.try_get() {
                                Some(listener_instance) => {
                                    let listener = listener_instance.read().unwrap().clone();
                                    let state = TunnelConnectionState::Completed;
                                    get_or_set_unique_bytebuf_tunnel_map(
                                        true,
                                        new_unique_bytebuf_tunnel.clone(),
                                        Some((pair.clone(), state, listener)),
                                    );
                                }
                                None => {
                                    log::error!("on_bybebuf_tunnel_select_duplex bytebuf error! please call set_sensor_type_tunnel_listener first!");
                                }
                            };
                        }
                    };
                    log::info!("on_bybebuf_tunnel_select_duplex find_unique_bytebuf_tunnel_by_tunnel_id end.")
                } else {
                    log::info!("on_bybebuf_tunnel_select_duplex unknow tunnel_data_type !")
                }
            }
        }

        0
    }

    /// received bytebuffer tunnel canceled as duplex rpc tunnel
    fn on_bybebuf_tunnel_cancel_duplex(
        &self,
        tunnel_id: String, // p2prpc tunnel of unique tunnel id
    ) -> i64 {
        log::info!("on_bybebuf_tunnel_cancel_duplex ");
        if let Ok(pair_result) = get_or_set_p2p_rtc_pair_map(false, tunnel_id.clone(), None) {
            if let Some((pair, state)) = pair_result {
                if pair.tunnel_data_type == TunnelDataTypeKind::BYTEBUF {
                    log::info!("on_bybebuf_tunnel_cancel_duplex select message tunnel as duplex.");
                    P2PRPCServiceContractImp::select_bytebuf_tunnel_duplex(tunnel_id.clone());
                } else {
                    log::info!("on_bybebuf_tunnel_cancel_duplex unknow tunnel_data_type !")
                }
            }
        }
        0
    }
}

struct EasyHTTPDServiceCallback {}
impl HttpdImpP2PCallBack for EasyHTTPDServiceCallback {
    fn on_httpd_p2p_ready(
        &self,
        web_root_host: String, // web root local host address.
    ) -> i64 {
        log::info!("on_httpd_p2p_ready host == {:?}", web_root_host);
        set_httpd_srv_started(true);
        notify_httpd_srv_started(web_root_host);
        0
    }
}

struct EasySGWebSocketCallBack {
    pub log_init_arg: OptionalLogIntArgs,
    pub srv_query_arg: OptionalServerQueryInitArgs,
}

impl SGWebSocketCallBack for EasySGWebSocketCallBack {
    fn on_sg_message(&self, data: String) -> i64 {
        log::info!("on_sg_message data = {}", data);
        0
    }

    fn on_sg_online(
        &self,
        client_id: String,
        turn_addr: String,
        turn_usr_name: String,
        turn_usr_pwd: String,
    ) -> i64 {
        log::info!(
            "on_sg_online client_id = {},turn_addr={},turn_usr_name={},turn_usr_pwd={}",
            client_id,
            turn_addr,
            turn_usr_name,
            turn_usr_pwd
        );

        set_websocketsrv_started(true);

        log::info!("websocketsrv started = {:?}", is_websocketsrv_started());
        let compare_cid = client_id.clone();
        if let Some(saved_client_id) = get_or_set_local_client_id(false, None) {
            if !saved_client_id.eq_ignore_ascii_case(&compare_cid) {
                log::error!("websocketsrv started saved client_id = {:?},not equal current client_id = {:?}.",saved_client_id,compare_cid);
            }
        }

        get_or_set_local_client_id(true, Some(client_id.clone()));

        if !is_p2pprc_started() {
            let p2p_rpc_init_arg = self.log_init_arg.to_owned();
            let p2p_rpc_query_arg = self.srv_query_arg.to_owned();
            let _ = P2PRPCServiceContractImp::init_p2prpc_sdk(
                p2p_rpc_init_arg.tag.unwrap().to_owned(),
                p2p_rpc_init_arg.log_level.unwrap().to_owned(),
            );
            std::thread::spawn(move || {
                let loop_p2p_query_arg = p2p_rpc_query_arg.clone();
                loop {
                    if is_manually_exited() {
                        log::warn!("p2prtm thread user manually exited");
                        break;
                    }
                    log::info!("p2p rpc start!");
                    let p2prpc_cb = Box::new(EasyP2PRPCServiceCallback {});
                    let mut srv_arg = loop_p2p_query_arg.clone();
                    let new_user_id = get_current_user_id();
                    if !srv_arg
                        .user_id
                        .clone()
                        .unwrap()
                        .eq_ignore_ascii_case(&new_user_id)
                    {
                        srv_arg.user_id = Some(new_user_id.to_owned());
                    }
                    let new_device_id = get_current_device_id();
                    if !srv_arg
                        .device_id
                        .clone()
                        .unwrap()
                        .eq_ignore_ascii_case(&new_device_id)
                    {
                        srv_arg.device_id = Some(new_device_id.to_owned());
                    }

                    // loop runing on turn server
                    let c_id = client_id.clone();
                    let t_addr = turn_addr.clone();
                    let t_uname = turn_usr_name.clone();
                    let t_upwd = turn_usr_pwd.clone();
                    let turn_config = P2PRPCConfig {
                        device_package: srv_arg.pkg_name.unwrap().to_owned(), // equal to ws init packagename arg;
                        device_id: srv_arg.device_id.unwrap().to_owned(), // equal to ws init device arg;
                        user_id: srv_arg.user_id.unwrap().to_owned(), // equal to ws init use arg;
                        client_id: c_id.to_owned(),
                        turn_uri: t_addr.to_owned(),
                        turn_usr: t_uname.to_owned(),
                        turn_pwd: t_upwd.to_owned(),
                        turn_ttl: 300,
                    };

                    if let Ok(turn_config_str) = serde_json::to_string(&turn_config) {
                        let res = P2PRPCServiceContractImp::loop_block_on_start_p2prpc_server(
                            srv_arg.host.unwrap().to_owned(),
                            srv_arg.turn_url_subgroup.unwrap().to_owned(),
                            srv_arg.app_key.unwrap().to_owned(),
                            srv_arg.app_secret.unwrap().to_owned(),
                            srv_arg.conig_save_dir_path.unwrap().to_owned(),
                            srv_arg.download_save_dir_path.unwrap().to_owned(),
                            false,
                            p2prpc_cb,
                            turn_config_str.to_owned(),
                        );
                        if res.code <= 0 {
                            log::error!(
                                "loop_block_on_start_p2prpc_server end .. res = {:?} !",
                                res.msg
                            );
                            std::thread::sleep(Duration::from_secs(3));
                            set_p2pprc_started(false);
                        }
                    };
                }
            });
            log::info!("p2p rpc callback join handle");
        } else {
            log::info!("p2p rpc already started!");
        }
        log::info!("end of on_ws_online...");
        0
    }

    fn on_sg_error(&self, err_msg: String) -> i64 {
        log::info!("on_sg_error err = {}", err_msg);
        0
    }

    fn on_sg_ping(&self) {
        log::info!("on_sg_ping");
    }

    fn on_sg_need_reconnect(&self, reason: String) -> i64 {
        log::info!("on_sg_need_reconnect reason = {}", reason);
        0
    }

    fn on_sg_send_reply(&self, send_event_id: String, is_ok: bool, reason: String) -> i64 {
        log::info!(
            "on_sg_send_reply send_event_id = {},is_ok ={},reason={}",
            send_event_id,
            is_ok,
            reason
        );
        0
    }
}

#[optional_struct]
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct LogIntArgs {
    pub tag: String,       // 日志工具初始化标签
    pub log_level: String, // 日志工具等级，分为：OFF，ERROR，WARN，INFO，DEBUG,TRACE几个等级
}
impl LogIntArgs {
    pub fn new(tag: String, log_level: String) -> Self {
        LogIntArgs { tag, log_level }
    }
}
impl OptionalLogIntArgs {
    pub fn default() -> Self {
        OptionalLogIntArgs {
            tag: Some("coolink".to_owned()),
            log_level: Some("INFO".to_owned()),
        }
    }
}

#[optional_struct]
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct ServerQueryInitArgs {
    pub host: String,                   // 酷连协议服务器域名
    pub sg_url_subgroup: String,        // 酷连协议信令网关subgroup请求地址
    pub turn_url_subgroup: String,      // 酷连协议穿透网关subgroup请求地址
    pub app_key: String,                // 酷连协议服务器接入AK
    pub app_secret: String,             // 酷连协议服务器接入密钥
    pub pkg_name: String,               //  酷连服务应用注册包名
    pub device_id: String,              //  酷连服务应用注册设备ID，可自动生成并存储即可
    pub user_id: String,                // 酷连服务应用注册用户ID，可对接各个第三方账号系统
    pub conig_save_dir_path: String,    // 本地配置文件存储路径
    pub download_save_dir_path: String, // 本地下载文件存储路径
}

impl ServerQueryInitArgs {
    pub fn new(
        host: String,
        sg_url_subgroup: String,
        turn_url_subgroup: String,
        app_key: String,
        app_secret: String,
        pkg_name: String,
        device_id: String,
        user_id: String,
        conig_save_dir_path: String,
        download_save_dir_path: String,
    ) -> Self {
        ServerQueryInitArgs {
            host,
            sg_url_subgroup,
            turn_url_subgroup,
            app_key,
            app_secret,
            pkg_name,
            device_id,
            user_id,
            conig_save_dir_path,
            download_save_dir_path,
        }
    }
}

pub fn local_default_saved_uid_and_did(config_saved_path: String) -> (String, String) {
    let config_store_path: PathBuf = config_saved_path.clone().into();
    let mut user_config = OptionalUserDefaultStorageConfig::load(
        &APP_INFO,
        USER_DEFAULT_LOCAL_CONFIG_FILE_NAME,
        &config_store_path,
    )
    .map_err(|e| {
        log::warn!("load user default config err: {}", e.to_string());
        e
    })
    .unwrap_or(OptionalUserDefaultStorageConfig::default());

    let new_default_user_map = user_config.user_default_ids_map.get_or_insert_default();

    let mut need_save = false;
    let default_user_id = if let Some(user_id) = new_default_user_map.get("user_id") {
        user_id.clone()
    } else {
        let random_id = ids::id_generate_xbit(8);
        let user_id = format!("tvos.uid-{}", random_id);
        new_default_user_map.insert("user_id".to_owned(), user_id.clone());
        need_save = true;
        user_id.to_owned()
    };

    let default_device_id = if let Some(device_id) = new_default_user_map.get("device_id") {
        device_id.clone()
    } else {
        let random_id = ids::id_generate_xbit(8);
        let device_id = format!("tvos.did-{}", random_id);
        new_default_user_map.insert("device_id".to_owned(), device_id.clone());
        need_save = true;
        device_id.to_owned()
    };

    if need_save {
        if let Err(e) = user_config.save(
            &APP_INFO,
            USER_DEFAULT_LOCAL_CONFIG_FILE_NAME,
            &config_store_path,
        ) {
            log::warn!("save user default config err: {:?}", e);
        } else {
            log::info!("save user default config Ok {:?}", &config_saved_path);
        }
    }
    (default_user_id, default_device_id)
}

impl OptionalServerQueryInitArgs {
    #[cfg(all(
        target_os = "linux",
        not(any(target_env = "musl", target_env = "ohos"))
    ))]
    pub fn default() -> Self {
        // 由于公开在开源社区，作者不承担该测试用信令网关服务的运行持续性和稳定性。
        // 内部测试服务的测试地址和访问key，性能受限，仅用于测试，且可能会由于测试环境的服务器关闭失效。
        // 商业环境如需使用，需要向TVOS组织申请正式环境参数，同时替换host、appkey、appsecret等切换为线上环境参数。
        let config_saved_path = "/home/rico/coolink/config/".to_owned();
        let (user_id, device_id) = local_default_saved_uid_and_did(config_saved_path.clone());
        OptionalServerQueryInitArgs {
            host: Some("http://obrr-api-test.skyworthiot.com".to_owned()),
            sg_url_subgroup: Some("/ws/get_lts_addrs".to_owned()),
            turn_url_subgroup: Some("/ws/get_turn_addrs".to_owned()),
            app_key: Some("12345678901234567890123456789012".to_owned()),
            app_secret: Some("d2969e51fe8ba0d4415270e8eb27ed62".to_owned()),
            pkg_name: Some("com.localhost.test".to_owned()),
            device_id: Some(device_id.to_owned()),
            user_id: Some(user_id.to_owned()),
            conig_save_dir_path: Some(config_saved_path.to_owned()),
            download_save_dir_path: Some("/home/rico/coolink/download/".to_owned()),
        }
    }
    #[cfg(all(target_os = "linux", any(target_env = "musl", target_env = "ohos")))]
    pub fn default() -> Self {
        // 由于公开在开源社区，作者不承担该测试用信令网关服务的运行持续性和稳定性。
        // 内部测试服务的测试地址和访问key，性能受限，仅用于测试，且可能会由于测试环境的服务器关闭失效。
        // 商业环境如需使用，需要向TVOS组织申请正式环境参数，同时替换host、appkey、appsecret等切换为线上环境参数。
        let config_saved_path = "/data/coolink/config/".to_owned();
        let (user_id, device_id) = local_default_saved_uid_and_did(config_saved_path.clone());
        OptionalServerQueryInitArgs {
            host: Some("http://obrr-api-test.skyworthiot.com".to_owned()),
            sg_url_subgroup: Some("/ws/get_lts_addrs".to_owned()),
            turn_url_subgroup: Some("/ws/get_turn_addrs".to_owned()),
            app_key: Some("12345678901234567890123456789012".to_owned()),
            app_secret: Some("d2969e51fe8ba0d4415270e8eb27ed62".to_owned()),
            pkg_name: Some("com.localhost.test".to_owned()),
            device_id: Some(device_id.to_owned()),
            user_id: Some(user_id.to_owned()),
            conig_save_dir_path: Some(config_saved_path.to_owned()),
            download_save_dir_path: Some("/data/coolink/download/".to_owned()),
        }
    }
}

#[optional_struct]
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct HttpdInitArgs {
    pub http_root_route_path: String,   // http服务本地根路由地址
    pub http_root_www_dir_path: String, // http服务根目录
    pub http_remote_route_path: String, // http服务P2PRPC文件服务路由地址
    pub http_remote_file_map_route_path: String, // http服务P2PRPC文件映射服务路由地址
    pub http_listen_host: String,
    pub http_listen_port: u32,
}

impl HttpdInitArgs {
    pub fn new(
        http_root_route_path: String,
        http_root_www_dir_path: String,
        http_remote_route_path: String,
        http_remote_file_map_route_path: String,
        http_listen_host: String,
        http_listen_port: u32,
    ) -> Self {
        HttpdInitArgs {
            http_root_route_path,
            http_root_www_dir_path,
            http_remote_route_path,
            http_remote_file_map_route_path,
            http_listen_host,
            http_listen_port,
        }
    }
}

impl OptionalHttpdInitArgs {
    #[cfg(all(
        target_os = "linux",
        not(any(target_env = "musl", target_env = "ohos"))
    ))]
    pub fn default() -> Self {
        OptionalHttpdInitArgs {
            http_root_route_path: Some("static".to_owned()),
            http_root_www_dir_path: Some("/home/rico/coolink/www/".to_owned()),
            http_remote_route_path: Some("remote".to_owned()),
            http_remote_file_map_route_path: Some("remote_map".to_owned()),
            http_listen_host: Some("0.0.0.0".to_owned()),
            http_listen_port: Some(0), //random port
        }
    }

    #[cfg(all(target_os = "linux", any(target_env = "musl", target_env = "ohos")))]
    pub fn default() -> Self {
        OptionalHttpdInitArgs {
            http_root_route_path: Some("static".to_owned()),
            http_root_www_dir_path: Some("/data/coolink/www/".to_owned()),
            http_remote_route_path: Some("remote".to_owned()),
            http_remote_file_map_route_path: Some("remote_map".to_owned()),
            http_listen_host: Some("0.0.0.0".to_owned()),
            http_listen_port: Some(0), //random port
        }
    }
}

pub struct CooLinkEasyApiImp {
    pub sdk_init_arg: OptionalLogIntArgs,
    pub srv_query_arg: OptionalServerQueryInitArgs,
    pub httpd_init_arg: OptionalHttpdInitArgs,
}

impl CooLinkEasyApiImp {
    pub fn new(
        log_init: LogIntArgs,
        server_init: ServerQueryInitArgs,
        httpd_init: HttpdInitArgs,
    ) -> Self {
        let mut sdk_init_arg = OptionalLogIntArgs::default();
        let mut srv_query_arg = OptionalServerQueryInitArgs::default();
        let mut httpd_init_arg = OptionalHttpdInitArgs::default();

        if !log_init.tag.is_empty() {
            sdk_init_arg.tag = Some(log_init.tag.to_owned());
        }
        if !log_init.log_level.is_empty() {
            sdk_init_arg.log_level = Some(log_init.log_level.to_owned());
        }
        if !server_init.host.is_empty() {
            srv_query_arg.host = Some(server_init.host);
        }
        if !server_init.sg_url_subgroup.is_empty() {
            srv_query_arg.sg_url_subgroup = Some(server_init.sg_url_subgroup);
        }
        if !server_init.turn_url_subgroup.is_empty() {
            srv_query_arg.turn_url_subgroup = Some(server_init.turn_url_subgroup);
        }
        if !server_init.app_key.is_empty() {
            srv_query_arg.app_key = Some(server_init.app_key);
        }
        if !server_init.app_secret.is_empty() {
            srv_query_arg.app_secret = Some(server_init.app_secret);
        }
        if !server_init.pkg_name.is_empty() {
            srv_query_arg.pkg_name = Some(server_init.pkg_name);
        }
        if !server_init.device_id.is_empty() {
            srv_query_arg.device_id = Some(server_init.device_id);
        }
        if !server_init.user_id.is_empty() {
            srv_query_arg.user_id = Some(server_init.user_id);
        }
        if !server_init.conig_save_dir_path.is_empty() {
            srv_query_arg.conig_save_dir_path = Some(server_init.conig_save_dir_path);
        }
        if !server_init.download_save_dir_path.is_empty() {
            srv_query_arg.download_save_dir_path = Some(server_init.download_save_dir_path);
        }
        if !httpd_init.http_root_route_path.is_empty() {
            httpd_init_arg.http_root_route_path = Some(httpd_init.http_root_route_path);
        }
        if !httpd_init.http_root_www_dir_path.is_empty() {
            httpd_init_arg.http_root_www_dir_path = Some(httpd_init.http_root_www_dir_path);
        }
        if !httpd_init.http_remote_route_path.is_empty() {
            httpd_init_arg.http_remote_route_path = Some(httpd_init.http_remote_route_path);
        }
        if !httpd_init.http_remote_file_map_route_path.is_empty() {
            httpd_init_arg.http_remote_file_map_route_path =
                Some(httpd_init.http_remote_file_map_route_path);
        }
        if !httpd_init.http_listen_host.is_empty() {
            httpd_init_arg.http_listen_host = Some(httpd_init.http_listen_host);
        }
        httpd_init_arg.http_listen_port = Some(httpd_init.http_listen_port);

        set_current_user_id(srv_query_arg.user_id.clone().unwrap());
        set_current_device_id(srv_query_arg.device_id.clone().unwrap());

        CooLinkEasyApiImp {
            sdk_init_arg,
            srv_query_arg,
            httpd_init_arg,
        }
    }

    pub fn start_websocket_srv(&self) -> JoinHandle<Result<(), anyhow::Error>> {
        let init_arg = self.sdk_init_arg.clone();
        let _ = SGWebSocketContractImp::init_sg_sdk(
            init_arg.tag.clone().unwrap().to_owned(),
            init_arg.log_level.clone().unwrap().to_owned(),
        );
        let loop_srv_arg = self.srv_query_arg.clone();
        let loop_init_arg = init_arg.clone();
        thread::spawn(move || {
            loop {
                if is_manually_exited() {
                    log::warn!("websocket loop: user manually exited");
                    break;
                }
                let mut srv_arg = loop_srv_arg.clone();
                let new_user_id = get_current_user_id();
                if !srv_arg
                    .user_id
                    .clone()
                    .unwrap()
                    .eq_ignore_ascii_case(&new_user_id)
                {
                    srv_arg.user_id = Some(new_user_id.to_owned());
                }
                let new_device_id = get_current_device_id();
                if !srv_arg
                    .device_id
                    .clone()
                    .unwrap()
                    .eq_ignore_ascii_case(&new_device_id)
                {
                    srv_arg.device_id = Some(new_device_id.to_owned());
                }

                let call_back_log_init_arg = loop_init_arg.clone();
                let call_back_query_srv_arg = srv_arg.clone();

                let websocket_cb: Box<EasySGWebSocketCallBack> =
                    Box::new(EasySGWebSocketCallBack {
                        log_init_arg: call_back_log_init_arg,
                        srv_query_arg: call_back_query_srv_arg,
                    });

                let res = SGWebSocketContractImp::loop_block_on_connect_sg(
                    // loop running on websocket srv
                    srv_arg.host.unwrap().to_owned(),
                    srv_arg.sg_url_subgroup.unwrap().to_owned(),
                    srv_arg.app_key.unwrap().to_owned(),
                    srv_arg.app_secret.unwrap().to_owned(),
                    srv_arg.pkg_name.unwrap().to_owned(), // equal to on ws online callback args RtmSdkServiceContractImp
                    srv_arg.device_id.unwrap().to_owned(), // equal to on ws online callback args
                    srv_arg.user_id.unwrap().to_owned(),  // equal to on ws online callback args
                    websocket_cb,
                    srv_arg.conig_save_dir_path.unwrap().to_owned(),
                );

                if res.code <= 0 {
                    log::error!("loop_block_on_connect_sg end .. res = {:?} !", res.msg);
                    std::thread::sleep(Duration::from_secs(3));
                    set_websocketsrv_started(false);
                }
                log::info!(
                    "loop_block_on_connect_sg break,websocket started = {:?}",
                    is_websocketsrv_started()
                );
            }
            Ok(())
        })
    }

    pub fn start_httpd_srv(&self) -> JoinHandle<Result<(), anyhow::Error>> {
        let loop_httpd_arg = self.httpd_init_arg.clone();
        thread::spawn(move || {
            loop {
                if is_manually_exited() {
                    log::warn!("httpd threaduser manually exited");
                    break;
                }

                let httpd_arg = loop_httpd_arg.clone();
                let httpd_cb: Box<EasyHTTPDServiceCallback> = Box::new(EasyHTTPDServiceCallback {});
                let res = HttpSrvContractImp::loop_block_on_httpd(
                    httpd_arg.http_root_route_path.unwrap().to_owned(),
                    // in openharmony,set root path = /data/www
                    httpd_arg.http_root_www_dir_path.unwrap().to_owned(),
                    httpd_arg.http_remote_route_path.unwrap().to_owned(),
                    httpd_arg
                        .http_remote_file_map_route_path
                        .unwrap()
                        .to_owned(),
                    httpd_arg.http_listen_host.unwrap().to_owned(),
                    httpd_arg.http_listen_port.unwrap().try_into().unwrap(),
                    httpd_cb,
                );
                if res.code <= 0 {
                    log::error!("loop_block_on_httpd end .. res = {:?} !", res.msg);
                    std::thread::sleep(Duration::from_secs(3));
                    set_httpd_srv_started(false);
                }
            }
            Ok(())
        })
    }

    pub fn reset_user_id(&self, new_user_id: String) {
        let current_user_id = get_current_user_id();
        if !new_user_id.is_empty() && !current_user_id.eq_ignore_ascii_case(&new_user_id) {
            set_current_user_id(new_user_id.to_owned());
            if is_p2pprc_started() {
                P2PRPCServiceContractImp::stop_p2prpc_server();
            }
            if is_websocketsrv_started() {
                SGWebSocketContractImp::disconnect_sg();
            }
            if is_httpd_srv_started() {
                HttpSrvContractImp::stop_httd();
            }
        }
    }

    pub fn reset_device_id(&self, new_device_id: String) {
        let current_device_id = get_current_device_id();
        if !new_device_id.is_empty() && !current_device_id.eq_ignore_ascii_case(&new_device_id) {
            set_current_device_id(new_device_id.to_owned());
            if is_p2pprc_started() {
                P2PRPCServiceContractImp::stop_p2prpc_server();
            }
            if is_websocketsrv_started() {
                SGWebSocketContractImp::disconnect_sg();
            }
            if is_httpd_srv_started() {
                HttpSrvContractImp::stop_httd();
            }
        }
    }

    fn connect_peer_as_controlling(
        &self,
        remote_client_id: String, // 对端ID
    ) -> i32 {
        if let Ok(remote_tunnels_vec) =
            find_unique_msg_tunnel_by_remote_c_id(remote_client_id.clone())
        {
            for (_, u_tunnel_opt) in remote_tunnels_vec.iter().enumerate() {
                if let Some(u_tunnel) = u_tunnel_opt {
                    let remot_client_id = u_tunnel.remote_id.clone();
                    if remot_client_id.eq_ignore_ascii_case(&remote_client_id) {
                        // 在map中找到这条连接ID
                        let ((pair, state, listener)) =
                            get_or_set_unique_msg_tunnel_map(false, u_tunnel.clone(), None)
                                .unwrap()
                                .unwrap();
                        match state {
                            TunnelConnectionState::Connected | TunnelConnectionState::Completed => {
                                listener.on_ready_to_send(u_tunnel.clone(), pair.is_controlling());
                            }
                            _ => {
                                log::info!(
                                    "connect_peer_as_controlling finded,but not ready to send"
                                );
                            }
                        }
                        return 1;
                    }
                }
            }
        }
        if let Some(saved_client_id) = get_or_set_local_client_id(false, None) {
            match UNIQUE_MSG_TUNNEL_CB.try_get() {
                Some(listener_instance) => {
                    log::info!("connect_peer_as_controlling in thread!");

                    let tunnel_id = ids::id_generate_xbit(6);
                    let new_pair = P2PRPCTunnelPair::new(
                        tunnel_id.clone(),
                        saved_client_id.clone(),
                        remote_client_id.clone(),
                        ControlKind::CONTROLLING,
                        TunnelDataTypeKind::LINKED,
                    );
                    get_or_set_p2p_rtc_pair_map(
                        true,
                        tunnel_id.clone(),
                        Some((new_pair.clone(), TunnelConnectionState::New)),
                    );
                    let new_txt_tunnel = UniqueTunnel {
                        tunnel_id: tunnel_id.clone(),
                        remote_id: remote_client_id.clone(),
                    };
                    let listener = listener_instance.read().unwrap().clone();
                    let state = TunnelConnectionState::New;
                    get_or_set_unique_msg_tunnel_map(
                        true,
                        new_txt_tunnel.clone(),
                        Some((new_pair.to_owned(), state, listener)),
                    );
                    thread::spawn(move || {
                        let res = P2PRPCServiceContractImp::block_on_make_p2prpc_tunnel(
                            saved_client_id.clone(),
                            remote_client_id.clone(),
                            tunnel_id.clone(),
                        );
                        if res.code <= 0 {
                            log::error!("block_on_make_p2prpc_tunnel end .. res = {:?} !", res.msg);
                            get_or_set_control_kind_map(true, tunnel_id.clone(), None);
                            get_or_set_p2p_rtc_pair_map(true, tunnel_id.clone(), None);
                            delete_unique_msg_tunnel_map(tunnel_id.clone());
                        }
                    });
                    return 0;
                }
                None => {
                    log::error!("connect_peer_as_controlling error! please call set_message_type_tunnel_listener first!");
                    return -1;
                }
            };
        } else {
            log::error!("connect_peer_as_controlling but not found clinet_id,return error!");
            return -1;
        }
    }

    fn open_sensor_as_controlling(
        &self,
        remote_client_id: String, // 对端ID
    ) -> i32 {
        if let Ok(remote_tunnels_vec) =
            find_unique_sensor_tunnel_by_remote_c_id(remote_client_id.clone())
        {
            for (_, u_tunnel_opt) in remote_tunnels_vec.iter().enumerate() {
                if let Some(u_tunnel) = u_tunnel_opt {
                    let unique_remote_id = u_tunnel.remote_id.clone();
                    if unique_remote_id.eq_ignore_ascii_case(&remote_client_id) {
                        // 在map中找到这条连接ID
                        let ((pair, state, listener)) =
                            get_or_set_unique_sensor_tunnel_map(false, u_tunnel.clone(), None)
                                .unwrap()
                                .unwrap();
                        match state {
                            TunnelConnectionState::Connected | TunnelConnectionState::Completed => {
                                listener.on_ready_to_send(u_tunnel.clone(), pair.is_controlling());
                            }
                            _ => {
                                log::info!(
                                    "open_sensor_as_controlling finded,but not ready to send"
                                );
                            }
                        }
                        return 1;
                    }
                }
            }
        }
        if let Some(saved_client_id) = get_or_set_local_client_id(false, None) {
            match UNIQUE_SENSOR_TUNNEL_CB.try_get() {
                Some(listener_instance) => {
                    log::info!("open_sensor_as_controlling in thread!");
                    let tunnel_id =
                        format!("{}{}", ids::id_generate_xbit(6), TUNNEL_ID_SENSOR_END_FIX);
                    let new_pair = P2PRPCTunnelPair::new(
                        tunnel_id.clone(),
                        saved_client_id.clone(),
                        remote_client_id.clone(),
                        ControlKind::CONTROLLING,
                        TunnelDataTypeKind::SENSOR,
                    );
                    get_or_set_p2p_rtc_pair_map(
                        true,
                        tunnel_id.clone(),
                        Some((new_pair.clone(), TunnelConnectionState::New)),
                    );
                    let new_txt_tunnel = UniqueSensorTunnel {
                        tunnel_id: tunnel_id.clone(),
                        remote_id: remote_client_id.clone(),
                    };
                    let listener = listener_instance.read().unwrap().clone();
                    let state = TunnelConnectionState::New;
                    get_or_set_unique_sensor_tunnel_map(
                        true,
                        new_txt_tunnel.clone(),
                        Some((new_pair.to_owned(), state, listener)),
                    );
                    thread::spawn(move || {
                        let res = P2PRPCServiceContractImp::block_on_make_p2prpc_tunnel(
                            saved_client_id.clone(),
                            remote_client_id.clone(),
                            tunnel_id.clone(),
                        );
                        if res.code <= 0 {
                            log::error!(
                                "block_on_make_p2prpc_tunnel sensor end .. res = {:?} !",
                                res.msg
                            );
                            get_or_set_control_kind_map(true, tunnel_id.clone(), None);
                            get_or_set_p2p_rtc_pair_map(true, tunnel_id.clone(), None);
                            delete_unique_sensor_tunnel_map(tunnel_id.clone());
                        }
                    });
                    return 0;
                }
                None => {
                    log::error!("open_sensor_as_controlling error! please call set_message_type_tunnel_listener first!");
                    return -1;
                }
            };
        } else {
            log::error!("open_sensor_as_controlling but not found clinet_id,return error!");
            return -1;
        }
    }

    fn start_byte_buf_channel_as_controlling(
        &self,
        remote_client_id: String, // 对端ID
    ) -> i32 {
        if let Ok(remote_tunnels_vec) =
            find_unique_bytebuf_tunnel_by_remote_c_id(remote_client_id.clone())
        {
            for (_, u_tunnel_opt) in remote_tunnels_vec.iter().enumerate() {
                if let Some(u_tunnel) = u_tunnel_opt {
                    let unique_remote_id = u_tunnel.remote_id.clone();
                    if unique_remote_id.eq_ignore_ascii_case(&remote_client_id) {
                        // 在map中找到这条连接ID
                        let ((pair, state, listener)) =
                            get_or_set_unique_bytebuf_tunnel_map(false, u_tunnel.clone(), None)
                                .unwrap()
                                .unwrap();
                        match state {
                            TunnelConnectionState::Connected | TunnelConnectionState::Completed => {
                                listener.on_ready_to_send(u_tunnel.clone(), pair.is_controlling());
                            }
                            _ => {
                                log::info!("start_byte_buf_channel_as_controlling finded,but not ready to send");
                            }
                        }
                        return 1;
                    }
                }
            }
        }
        if let Some(saved_client_id) = get_or_set_local_client_id(false, None) {
            match UNIQUE_BYTEBUF_TUNNEL_CB.try_get() {
                Some(listener_instance) => {
                    log::info!("start_byte_buf_channel_as_controlling in thread!");
                    let tunnel_id =
                        format!("{}{}", ids::id_generate_xbit(6), TUNNEL_ID_BYTEBUF_END_FIX);
                    let new_pair = P2PRPCTunnelPair::new(
                        tunnel_id.clone(),
                        saved_client_id.clone(),
                        remote_client_id.clone(),
                        ControlKind::CONTROLLING,
                        TunnelDataTypeKind::BYTEBUF,
                    );
                    get_or_set_p2p_rtc_pair_map(
                        true,
                        tunnel_id.clone(),
                        Some((new_pair.clone(), TunnelConnectionState::New)),
                    );
                    let new_txt_tunnel = UniqueByteBufTunnel {
                        tunnel_id: tunnel_id.clone(),
                        remote_id: remote_client_id.clone(),
                    };
                    let listener = listener_instance.read().unwrap().clone();
                    let state = TunnelConnectionState::New;
                    get_or_set_unique_bytebuf_tunnel_map(
                        true,
                        new_txt_tunnel.clone(),
                        Some((new_pair.to_owned(), state, listener)),
                    );
                    thread::spawn(move || {
                        let res = P2PRPCServiceContractImp::block_on_make_p2prpc_tunnel(
                            saved_client_id.clone(),
                            remote_client_id.clone(),
                            tunnel_id.clone(),
                        );
                        if res.code <= 0 {
                            log::error!(
                                "block_on_make_p2prpc_tunnel bytebuf end .. res = {:?} !",
                                res.msg
                            );
                            get_or_set_control_kind_map(true, tunnel_id.clone(), None);
                            get_or_set_p2p_rtc_pair_map(true, tunnel_id.clone(), None);
                            delete_unique_bytebuf_tunnel_map(tunnel_id.clone());
                        }
                    });
                    return 0;
                }
                None => {
                    log::error!("start_byte_buf_channel_as_controlling error! please call set_message_type_tunnel_listener first!");
                    return -1;
                }
            };
        } else {
            log::error!(
                "start_byte_buf_channel_as_controlling but not found clinet_id,return error!"
            );
            return -1;
        }
    }

    fn is_message_tunnel_ready(
        &self,
        remote_client_id: String, // 对端ID
    ) -> bool {
        let mut is_ready = false;
        if let Ok(remote_tunnels_vec) =
            find_unique_msg_tunnel_by_remote_c_id(remote_client_id.clone())
        {
            for (_, u_tunnel_opt) in remote_tunnels_vec.iter().enumerate() {
                if let Some(u_tunnel) = u_tunnel_opt {
                    let unique_remote_id = u_tunnel.remote_id.clone();
                    if unique_remote_id.eq_ignore_ascii_case(&remote_client_id) {
                        let ((_, state, _)) =
                            get_or_set_unique_msg_tunnel_map(false, u_tunnel.clone(), None)
                                .unwrap()
                                .unwrap();
                        match state {
                            TunnelConnectionState::Connected | TunnelConnectionState::Completed => {
                                is_ready = true;
                            }
                            _ => {
                                log::info!("is_message_tunnel_ready = false");
                            }
                        }
                        break;
                    }
                }
            }
        }

        is_ready
    }

    fn is_sensor_tunnel_ready(
        &self,
        remote_client_id: String, // 对端ID
    ) -> bool {
        let mut is_ready = false;
        if let Ok(remote_tunnels_vec) =
            find_unique_sensor_tunnel_by_remote_c_id(remote_client_id.clone())
        {
            for (_, u_tunnel_opt) in remote_tunnels_vec.iter().enumerate() {
                if let Some(u_tunnel) = u_tunnel_opt {
                    let unique_remote_id = u_tunnel.remote_id.clone();
                    if unique_remote_id.eq_ignore_ascii_case(&remote_client_id) {
                        let ((_, state, _)) =
                            get_or_set_unique_sensor_tunnel_map(false, u_tunnel.clone(), None)
                                .unwrap()
                                .unwrap();
                        match state {
                            TunnelConnectionState::Connected | TunnelConnectionState::Completed => {
                                is_ready = true;
                            }
                            _ => {
                                log::info!("is_sensor_tunnel_ready = false");
                            }
                        }
                        break;
                    }
                }
            }
        }

        is_ready
    }

    fn is_bytebuf_tunnel_ready(
        &self,
        remote_client_id: String, // 对端ID
    ) -> bool {
        let mut is_ready = false;
        if let Ok(remote_tunnels_vec) =
            find_unique_bytebuf_tunnel_by_remote_c_id(remote_client_id.clone())
        {
            for (_, u_tunnel_opt) in remote_tunnels_vec.iter().enumerate() {
                if let Some(u_tunnel) = u_tunnel_opt {
                    let unique_remote_id = u_tunnel.remote_id.clone();
                    if unique_remote_id.eq_ignore_ascii_case(&remote_client_id) {
                        let ((_, state, _)) =
                            get_or_set_unique_bytebuf_tunnel_map(false, u_tunnel.clone(), None)
                                .unwrap()
                                .unwrap();
                        match state {
                            TunnelConnectionState::Connected | TunnelConnectionState::Completed => {
                                is_ready = true;
                            }
                            _ => {
                                log::info!("is_bytebuf_tunnel_ready = false");
                            }
                        }
                        break;
                    }
                }
            }
        }

        is_ready
    }

    fn destroy_tunnel(
        &self,
        remote_client_id: String, // 对端ID
    ) -> bool {
        if let Ok(remote_tunnels_vec) =
            find_unique_msg_tunnel_by_remote_c_id(remote_client_id.clone())
        {
            for (_, u_tunnel_opt) in remote_tunnels_vec.iter().enumerate() {
                if let Some(u_tunnel) = u_tunnel_opt {
                    let remote_unique_tunnel_id = u_tunnel.tunnel_id.clone();
                    log::warn!(
                        "destroy_tunnel message tunnel: {:?}",
                        remote_unique_tunnel_id.clone()
                    );
                    P2PRPCServiceContractImp::drop_p2prpc_tunnel(
                        remote_unique_tunnel_id.to_owned(),
                    );
                    get_or_set_unique_msg_tunnel_map(true, u_tunnel.clone(), None);
                }
            }
        }

        if let Ok(remote_tunnels_vec) =
            find_unique_sensor_tunnel_by_remote_c_id(remote_client_id.clone())
        {
            for (_, u_tunnel_opt) in remote_tunnels_vec.iter().enumerate() {
                if let Some(u_tunnel) = u_tunnel_opt {
                    let remote_unique_tunnel_id = u_tunnel.tunnel_id.clone();
                    log::warn!(
                        "destroy_tunnel sensor tunnel: {:?}",
                        remote_unique_tunnel_id.clone()
                    );
                    P2PRPCServiceContractImp::drop_p2prpc_tunnel(
                        remote_unique_tunnel_id.to_owned(),
                    );
                    get_or_set_unique_sensor_tunnel_map(true, u_tunnel.clone(), None);
                }
            }
        }

        if let Ok(remote_tunnels_vec) =
            find_unique_bytebuf_tunnel_by_remote_c_id(remote_client_id.clone())
        {
            for (_, u_tunnel_opt) in remote_tunnels_vec.iter().enumerate() {
                if let Some(u_tunnel) = u_tunnel_opt {
                    let remote_unique_tunnel_id = u_tunnel.tunnel_id.clone();
                    log::warn!(
                        "destroy_tunnel bytebuffer tunnel: {:?}",
                        remote_unique_tunnel_id.clone()
                    );
                    P2PRPCServiceContractImp::drop_p2prpc_tunnel(
                        remote_unique_tunnel_id.to_owned(),
                    );
                    get_or_set_unique_bytebuf_tunnel_map(true, u_tunnel.clone(), None);
                }
            }
        }

        true
    }

    fn get_connected_devices(&self) -> Vec<Option<UniqueTunnel>> {
        let map = UNIQUEMSGTUNNEL_MAP.read().unwrap();
        let mut unique_tunnel_opt_vec: Vec<Option<UniqueTunnel>> = vec![];

        for (u_tunnel, _) in map.iter() {
            unique_tunnel_opt_vec.push(Some(u_tunnel.clone()))
        }
        unique_tunnel_opt_vec
    }

    fn send_message(
        &self,
        remote_client_id: String, // 对端设备ID

        message_id: String,    // 消息ID
        message_event: String, // 消息名
        message: String,       // 消息
    ) -> bool {
        let mut tunnel_id = "".to_string();
        if let Ok(remote_tunnels_vec) =
            find_unique_msg_tunnel_by_remote_c_id(remote_client_id.clone())
        {
            for (_, u_tunnel_opt) in remote_tunnels_vec.iter().enumerate() {
                if let Some(u_tunnel) = u_tunnel_opt {
                    tunnel_id = u_tunnel.tunnel_id.clone();
                    log::warn!("send_message message find tunnel: {:?}", tunnel_id.clone());
                    break;
                }
            }
        }
        if tunnel_id.is_empty() {
            return false;
        } else {
            let res = P2PRPCServiceContractImp::send_text_message(
                tunnel_id,
                message_id,
                message_event,
                message,
                false,
            );
            if res.code == 0 {
                return true;
            } else {
                log::info!("send_message failure,reason is :{:?}", res.msg);
                return false;
            }
        }
    }

    fn send_sensor(
        &self,
        remote_client_id: String, // 对端设备ID

        message_id: String,   // 消息ID
        sensor_event: String, // sensor消息名称
        sensor_data: String,  // sensor消息
    ) -> bool {
        let mut tunnel_id = "".to_string();
        if let Ok(remote_tunnels_vec) =
            find_unique_sensor_tunnel_by_remote_c_id(remote_client_id.clone())
        {
            for (_, u_tunnel_opt) in remote_tunnels_vec.iter().enumerate() {
                if let Some(u_tunnel) = u_tunnel_opt {
                    tunnel_id = u_tunnel.tunnel_id.clone();
                    log::warn!("send_sensor message find tunnel: {:?}", tunnel_id.clone());
                    break;
                }
            }
        }
        if tunnel_id.is_empty() {
            return false;
        } else {
            let res = P2PRPCServiceContractImp::send_text_message(
                tunnel_id,
                message_id,
                sensor_event,
                sensor_data,
                false,
            );
            if res.code == 0 {
                return true;
            } else {
                log::info!("send_message failure,reason is :{:?}", res.msg);
                return false;
            }
        }
    }

    fn send_bytebuf(
        &self,
        remote_client_id: String, // 对端设备ID

        message_id: String,         // 消息ID
        can_lost: bool,             // 能否丢包
        video_buf: Option<Vec<u8>>, // 视频数据
        audio_buf: Option<Vec<u8>>, // 音频数据
    ) -> bool {
        let mut tunnel_id = "".to_string();
        if let Ok(remote_tunnels_vec) =
            find_unique_bytebuf_tunnel_by_remote_c_id(remote_client_id.clone())
        {
            for (_, u_tunnel_opt) in remote_tunnels_vec.iter().enumerate() {
                if let Some(u_tunnel) = u_tunnel_opt {
                    tunnel_id = u_tunnel.tunnel_id.clone();
                    log::warn!("send_bytebuf message find tunnel: {:?}", tunnel_id.clone());
                    break;
                }
            }
        }
        if tunnel_id.is_empty() {
            return false;
        } else {
            let res = P2PRPCServiceContractImp::send_bytebuffer(
                tunnel_id,
                message_id,
                video_buf.unwrap_or(vec![]),
                audio_buf.unwrap_or(vec![]),
                can_lost,
            );
            if res.code == 0 {
                return true;
            } else {
                log::info!("send_message failure,reason is :{:?}", res.msg);
                return false;
            }
        }
    }

    fn generate_p2prpc_file_mapping(
        &self,
        file_path: String, // 本地文件的绝对路径
    ) -> Result<String, anyhow::Error> {
        let res = P2PRPCServiceContractImp::soft_link_file_mapping(1, file_path);
        if res.code == 0 {
            return Ok(res.msg);
        } else {
            anyhow::bail!(anyhow::anyhow!(res.msg));
        }
    }

    fn is_symbol_link_file(&self, mapping_url: String) -> bool {
        if mapping_url.starts_with("_sym_") {
            true
        } else {
            false
        }
    }

    fn compose_httpd_url(
        &self,
        remote_client_id: String, // 对方的设备id
        mapping_url: String,      // 映射的url名字
    ) -> Option<String> {
        let res = HttpSrvContractImp::get_httpd_host();
        if res.code == 0 {
            let compose_httpd_url = format!(
                "{}/{}/{}/{}",
                res.msg,
                self.httpd_init_arg
                    .http_remote_file_map_route_path
                    .clone()
                    .unwrap()
                    .to_owned(),
                remote_client_id,
                mapping_url
            );
            log::info!("compose_httpd_url = {:?}", compose_httpd_url);
            Some(compose_httpd_url)
        } else {
            None
        }
    }

    fn compose_httpd_prefer_lan_url(
        &self,
        remote_client_id: String, // 对方的设备id
        mapping_url: String,
    ) -> Option<String> {
        let res = HttpSrvContractImp::get_httpd_host_prefer_lan(
            remote_client_id.clone(),
            self.httpd_init_arg.http_root_route_path.clone().unwrap(),
            mapping_url.clone(),
        );
        if res.code == 0 {
            if let Ok(file_in_lan) = serde_json::from_str::<FileInLanHttpAddress>(&res.msg) {
                if file_in_lan.is_in_lan {
                    log::info!("compose_httpd_prefer_lan_url = {:?}", file_in_lan.host);
                    Some(file_in_lan.host)
                } else {
                    self.compose_httpd_url(remote_client_id, mapping_url)
                }
            } else {
                None
            }
        } else {
            None
        }
    }

    fn get_p2prpc_file_mapping_to_http(
        &self,
        remote_client_id: String, // 对方的设备id
        mapping_url: String,      // 映射的url名字
    ) -> Result<String, anyhow::Error> {
        let mut tunnel_id = "".to_string();
        if let Ok(remote_tunnels_vec) =
            find_unique_bytebuf_tunnel_by_remote_c_id(remote_client_id.clone())
        {
            for (_, u_tunnel_opt) in remote_tunnels_vec.iter().enumerate() {
                if let Some(u_tunnel) = u_tunnel_opt {
                    tunnel_id = u_tunnel.tunnel_id.clone();
                    log::warn!(
                        "get_p2prpc_file_mapping_to_http message find tunnel: {:?}",
                        tunnel_id.clone()
                    );
                    break;
                }
            }
        }
        if tunnel_id.is_empty() {
            anyhow::bail!("tunnel id is empty.");
        } else {
            if self.is_symbol_link_file(mapping_url.clone()) {
                if let Some(compose_url) =
                    self.compose_httpd_prefer_lan_url(remote_client_id, mapping_url)
                {
                    return Ok(compose_url);
                } else {
                    anyhow::bail!("compose_httpd_prefer_lan_url null");
                }
            } else {
                if let Some(compose_url) = self.compose_httpd_url(remote_client_id, mapping_url) {
                    return Ok(compose_url);
                } else {
                    anyhow::bail!("compose_httpd_url null");
                }
            }
        }
    }

    fn query_online_from_gateway(
        &self,
        client_id: String,                                 // 接入的client_id
        query_callback: Arc<Box<TempQueryOnLineCallback>>, // 接口请求回调
    ) -> Result<(), anyhow::Error> {
        let client_id = client_id.clone();
        let query_callback = query_callback.clone();
        let srv_query_arg = self.srv_query_arg.clone();
        std::thread::spawn(move || {
            let res = SGWebSocketContractImp::block_query_sg_is_online(
                srv_query_arg.host.unwrap().to_owned(),
                "/ws/is_online".to_string(),
                srv_query_arg.app_key.unwrap().to_owned(),
                srv_query_arg.app_secret.unwrap().to_owned(),
                client_id.clone(),
            );
            if res.code == 0 && res.msg.eq_ignore_ascii_case("1") {
                //success:bool,client_id:String,online:bool
                query_callback.on_client_id_online(true, client_id.to_owned(), true);
            } else {
                query_callback.on_client_id_online(false, client_id.to_owned(), false);
            }
        });
        Ok(())
    }

    fn is_in_same_host(
        &self,
        remote_client_id: String, //远端client_id
    ) -> bool {
        let mut tunnel_id = "".to_string();
        if let Ok(remote_tunnels_vec) =
            find_unique_msg_tunnel_by_remote_c_id(remote_client_id.clone())
        {
            for (_, u_tunnel_opt) in remote_tunnels_vec.iter().enumerate() {
                if let Some(u_tunnel) = u_tunnel_opt {
                    tunnel_id = u_tunnel.tunnel_id.clone();
                    log::warn!("send_message message find tunnel: {:?}", tunnel_id.clone());
                    break;
                }
            }
        }
        if tunnel_id.is_empty() {
            return false;
        } else {
            let res =
                P2PRPCServiceContractImp::is_p2prpc_remote_in_lan(tunnel_id, remote_client_id);
            if res.code == 0 {
                return true;
            } else {
                return false;
            }
        }
    }
}

// set easy sdk instance only once
fn set_easy_sdk_instance(coolink_easy_api: CoolinkSDKEasyApi) -> bool {
    // if try get success from state::Storage
    let ok = if let Some(_) = EASY_API_INSTANCE.try_get() {
        // if sdk instance call many times in many thread and this api will return false.
        log::warn!("easy api sdk instance already inited!");
        false
    } else {
        // set new connect_state
        EASY_API_INSTANCE.set(RwLock::new(coolink_easy_api));
        true
    };
    ok
}

fn set_unique_msg_tunnel_callback(listener: Box<dyn IUniqueMessageTunnelListener>) -> bool {
    // if try get success from state::Storage
    let ok = if let Some(_) = UNIQUE_MSG_TUNNEL_CB.try_get() {
        // if sdk instance call many times in many thread and this api will return false.
        log::warn!("unique message tunnel callback inited!");
        false
    } else {
        // set new connect_state
        UNIQUE_MSG_TUNNEL_CB.set(RwLock::new(Arc::new(listener)));
        true
    };
    ok
}

fn set_sensor_msg_tunnel_callback(listener: Box<dyn ISensorTunnelListener>) -> bool {
    // if try get success from state::Storage
    let ok = if let Some(_) = UNIQUE_SENSOR_TUNNEL_CB.try_get() {
        // if sdk instance call many times in many thread and this api will return false.
        log::warn!("unique sensor tunnel callback inited!");
        false
    } else {
        // set new connect_state
        UNIQUE_SENSOR_TUNNEL_CB.set(RwLock::new(Arc::new(listener)));
        true
    };
    ok
}

fn set_bytebuf_msg_tunnel_callback(listener: Box<dyn IByteBufListener>) -> bool {
    // if try get success from state::Storage
    let ok = if let Some(_) = UNIQUE_BYTEBUF_TUNNEL_CB.try_get() {
        // if sdk instance call many times in many thread and this api will return false.
        log::warn!("unique bytebuf tunnel callback inited!");
        false
    } else {
        // set new connect_state
        UNIQUE_BYTEBUF_TUNNEL_CB.set(RwLock::new(Arc::new(listener)));
        true
    };
    ok
}

pub struct CoolinkSDKEasyApi {
    easy_instance: CooLinkEasyApiImp,
    // msg id ,
    send_msg_result_map: RwLock<HashMap<String, Arc<Box<TempSendMsgResultCallback>>>>,
    send_sensor_result_map: RwLock<HashMap<String, Arc<Box<TempSendSensorResultCallback>>>>,
    send_bytebuf_result_map: RwLock<HashMap<String, Arc<Box<TempSendByteBufResultCallback>>>>,

    ws_handler: JoinHandle<Result<(), anyhow::Error>>,
    httpd_handelr: JoinHandle<Result<(), anyhow::Error>>,
}

fn callback_msg_tunnel_message(
    msg_id: String,  // unique msgid sent
    is_ok: bool,     // message sent is ok
    err_msg: String, // if is_ok == false,error message print
) {
    match EASY_API_INSTANCE.try_get() {
        Some(instance) => {
            instance
                .read()
                .unwrap()
                .drain_send_msg_result_callback(is_ok, msg_id, err_msg);
        }
        None => {
            log::error!("error! please call init_easy_sdk first!");
        }
    };
}

fn callback_sensor_tunnel_message(
    msg_id: String,  // unique msgid sent
    is_ok: bool,     // message sent is ok
    err_msg: String, // if is_ok == false,error message print
) {
    match EASY_API_INSTANCE.try_get() {
        Some(instance) => {
            instance
                .read()
                .unwrap()
                .drain_send_sensor_result_callback(is_ok, msg_id, err_msg);
        }
        None => {
            log::error!("error! please call init_easy_sdk first!");
        }
    };
}

fn callback_bytebuf_tunnel_message(
    msg_id: String,  // unique msgid sent
    is_ok: bool,     // message sent is ok
    err_msg: String, // if is_ok == false,error message print
) {
    match EASY_API_INSTANCE.try_get() {
        Some(instance) => {
            instance
                .read()
                .unwrap()
                .drain_send_bytebuf_result_callback(is_ok, msg_id, err_msg);
        }
        None => {
            log::error!("error! please call init_easy_sdk first!");
        }
    };
}

struct TempSendMsgResultCallback {
    pub msg_send_result_listener: Box<dyn ISendMessageResult>,
}

impl TempSendMsgResultCallback {
    fn on_result(&self, success: bool, msg_id: String, err_msg: String) {
        self.msg_send_result_listener
            .on_result(success, msg_id, err_msg);
    }
}

struct TempSendSensorResultCallback {
    pub sensor_send_result_listener: Box<dyn ISendSensorResult>,
}

impl TempSendSensorResultCallback {
    fn on_result(&self, success: bool, msg_id: String, err_msg: String) {
        self.sensor_send_result_listener
            .on_result(success, msg_id, err_msg);
    }
}

struct TempSendByteBufResultCallback {
    pub bytebuf_send_result_listener: Box<dyn ISendByteBufResult>,
}

impl TempSendByteBufResultCallback {
    fn on_result(&self, success: bool, msg_id: String, err_msg: String) {
        self.bytebuf_send_result_listener
            .on_result(success, msg_id, err_msg);
    }
}

struct TempQueryOnLineCallback {
    pub temp_query_cb_listener: Box<dyn IQueryOnline>,
}

impl TempQueryOnLineCallback {
    fn on_client_id_online(&self, success: bool, client_id: String, online: bool) {
        self.temp_query_cb_listener
            .on_client_id_online(success, client_id, online);
    }
}

impl CoolinkSDKEasyApi {
    fn new(
        log_init: LogIntArgs,
        server_init: ServerQueryInitArgs,
        httpd_init: HttpdInitArgs,
    ) -> Self {
        let easy_instance = CooLinkEasyApiImp::new(log_init, server_init, httpd_init);

        set_websocketsrv_started(false);
        set_httpd_srv_started(false);
        set_p2pprc_started(false);
        IS_MANUALLY_EXITED.set(RwLock::new(false));
        let ws_handler = easy_instance.start_websocket_srv();
        let httpd_handelr = easy_instance.start_httpd_srv();

        let send_msg_result_map = RwLock::new(HashMap::new());
        let send_sensor_result_map = RwLock::new(HashMap::new());
        let send_bytebuf_result_map = RwLock::new(HashMap::new());

        CoolinkSDKEasyApi {
            easy_instance,
            ws_handler,
            httpd_handelr,
            send_msg_result_map,
            send_sensor_result_map,
            send_bytebuf_result_map,
        }
    }

    pub fn shut_down() {
        if !is_manually_exited() {
            IS_MANUALLY_EXITED.set(RwLock::new(true));
            HttpSrvContractImp::stop_httd();
            P2PRPCServiceContractImp::stop_p2prpc_server();
            SGWebSocketContractImp::disconnect_sg();

            let three_seconds = std::time::Duration::from_secs(3);
            thread::sleep(three_seconds);
        } else {
            log::error!("manually shut_down~");
        }
    }

    fn set_message_type_tunnel_listener(&self, listener: Box<dyn IUniqueMessageTunnelListener>) {
        if !set_unique_msg_tunnel_callback(listener) {
            log::warn!("already set IUniqueMessageTunnelListener.");
        }
    }

    fn set_sensor_type_tunnel_listener(&self, listener: Box<dyn ISensorTunnelListener>) {
        if !set_sensor_msg_tunnel_callback(listener) {
            log::warn!("already set ISensorTunnelListener.");
        }
    }

    fn set_bytebuf_type_tunnel_listener(&self, listener: Box<dyn IByteBufListener>) {
        if !set_bytebuf_msg_tunnel_callback(listener) {
            log::warn!("already set IByteBufListener.");
        }
    }

    fn re_init_user_id(
        &self,
        user_id: String, // 新的用户id
    ) {
        self.easy_instance.reset_user_id(user_id.to_owned());
    }

    fn re_init_device_id(
        &self,
        device_id: String, // 新的设备id
    ) {
        self.easy_instance.reset_device_id(device_id.to_owned());
    }

    // 返回int： 1 代表连接已经存在，默认是成功的，0代表此次成功，-1代表有错误发生
    fn connect_peer_as_controlling(
        &self,
        remote_client_id: String, // 对端ID
    ) -> i32 {
        self.easy_instance
            .connect_peer_as_controlling(remote_client_id.to_owned())
    }

    fn open_sensor_as_controlling(
        &self,
        remote_client_id: String, // 对端ID
    ) -> i32 {
        self.easy_instance
            .open_sensor_as_controlling(remote_client_id.to_owned())
    }

    fn start_byte_buf_channel_as_controlling(
        &self,
        remote_client_id: String, // 对端ID
    ) -> i32 {
        self.easy_instance
            .start_byte_buf_channel_as_controlling(remote_client_id.to_owned())
    }

    fn is_message_tunnel_ready(
        &self,
        remote_client_id: String, // 对端ID
    ) -> bool {
        self.easy_instance.is_message_tunnel_ready(remote_client_id)
    }

    fn is_sensor_tunnel_ready(
        &self,
        remote_client_id: String, // 对端ID
    ) -> bool {
        self.easy_instance.is_sensor_tunnel_ready(remote_client_id)
    }

    fn is_bytebuf_tunnel_ready(
        &self,
        remote_client_id: String, // 对端ID
    ) -> bool {
        self.easy_instance.is_bytebuf_tunnel_ready(remote_client_id)
    }

    fn destroy_tunnel(
        &self,
        remote_client_id: String, // 对端ID
    ) -> bool {
        self.easy_instance.destroy_tunnel(remote_client_id)
    }

    fn get_connected_devices(&self) -> Vec<Option<UniqueTunnel>> {
        self.easy_instance.get_connected_devices()
    }

    fn put_send_msg_result_map(
        &self,
        msg_id: String,
        listener: Arc<Box<TempSendMsgResultCallback>>,
    ) {
        let mut map = self.send_msg_result_map.write().unwrap();
        map.insert(msg_id.to_owned(), listener);
    }

    fn drain_send_msg_result_callback(&self, result: bool, msg_id: String, err_msg: String) {
        let mut map = self.send_msg_result_map.write().unwrap();
        if let Some(listener) = map.remove(&msg_id) {
            listener.on_result(result, msg_id, err_msg);
            log::info!(
                "drain_send_msg_result_callback strong count = {:?}",
                Arc::strong_count(&listener),
            );
        }
    }

    fn drain_all_send_msg_result_callbacks(&self) {
        let mut map = self.send_msg_result_map.write().unwrap();
        for (msg_id, listener) in map.drain().take(1) {
            listener.on_result(
                false,
                msg_id.clone(),
                "message tunnel drop error".to_string(),
            );
        }
    }

    fn put_send_sensor_result_map(
        &self,
        msg_id: String,
        listener: Arc<Box<TempSendSensorResultCallback>>,
    ) {
        let mut map = self.send_sensor_result_map.write().unwrap();
        map.insert(msg_id.to_owned(), listener);
    }

    fn drain_send_sensor_result_callback(&self, result: bool, msg_id: String, err_msg: String) {
        let mut map = self.send_sensor_result_map.write().unwrap();
        if let Some(listener) = map.remove(&msg_id) {
            listener.on_result(result, msg_id, err_msg);
            log::info!(
                "drain_send_sensor_result_callback strong count = {:?}",
                Arc::strong_count(&listener),
            );
        }
    }

    fn drain_all_send_sensor_result_callbacks(&self) {
        let mut map = self.send_sensor_result_map.write().unwrap();
        for (msg_id, listener) in map.drain().take(1) {
            listener.on_result(
                false,
                msg_id.clone(),
                "sensor tunnel drop error".to_string(),
            );
        }
    }

    fn put_send_bytebuf_result_map(
        &self,
        msg_id: String,
        listener: Arc<Box<TempSendByteBufResultCallback>>,
    ) {
        let mut map = self.send_bytebuf_result_map.write().unwrap();
        map.insert(msg_id.to_owned(), listener);
    }

    fn drain_send_bytebuf_result_callback(&self, result: bool, msg_id: String, err_msg: String) {
        let mut map = self.send_bytebuf_result_map.write().unwrap();
        if let Some(listener) = map.remove(&msg_id) {
            listener.on_result(result, msg_id, err_msg);
            log::info!(
                "drain_send_bytebuf_result_callback strong count = {:?}",
                Arc::strong_count(&listener),
            );
        }
    }

    fn drain_all_send_bytebuf_result_callbacks(&self) {
        let mut map = self.send_bytebuf_result_map.write().unwrap();
        for (msg_id, listener) in map.drain().take(1) {
            listener.on_result(
                false,
                msg_id.clone(),
                "bytebuf tunnel drop error".to_string(),
            );
        }
    }

    fn send_message(
        &self,
        temp_send_result: Arc<Box<TempSendMsgResultCallback>>, // 回调指针
        remote_client_id: String,                              // 对端设备ID
        message_event: String,                                 // 消息名
        message: String,                                       // 消息
    ) -> Result<String, anyhow::Error> {
        let msg_id = ids::id_generate_xbit(10);
        self.put_send_msg_result_map(msg_id.clone(), temp_send_result);
        let res = self.easy_instance.send_message(
            remote_client_id,
            msg_id.clone(),
            message_event,
            message,
        );
        if res {
            return Ok(msg_id.to_owned());
        } else {
            let err_msg = "send_message transport error.".to_string();
            self.drain_send_msg_result_callback(false, msg_id.to_owned(), err_msg.clone());
            anyhow::bail!(err_msg);
        }
    }

    fn broadcast_message(
        &self,
        temp_send_result: Arc<Box<TempSendMsgResultCallback>>,
        message_event: String, // 消息名
        message: String,       // 消息
    ) -> Vec<Result<(String, String), anyhow::Error>> {
        let mut tunnel_vec = vec![];
        let map = UNIQUEMSGTUNNEL_MAP.read().unwrap();
        for (u_tunnel, _) in map.iter() {
            tunnel_vec.push(u_tunnel);
        }

        let mut result_vec = vec![];
        for elem in tunnel_vec.iter() {
            let iter_temp_listener = temp_send_result.clone();
            if let Ok(msgid) = self.send_message(
                iter_temp_listener,
                elem.remote_id.clone(),
                message_event.clone(),
                message.clone(),
            ) {
                result_vec.push(Ok((elem.remote_id.clone(), msgid)))
            }
        }
        result_vec
    }

    fn send_sensor(
        &self,
        temp_send_result: Arc<Box<TempSendSensorResultCallback>>, // 回调指针
        remote_client_id: String,                                 // 对端设备ID
        sensor_event: String,                                     // sensor消息名称
        sensor_data: String,                                      // sensor消息
    ) -> Result<String, anyhow::Error> {
        let msg_id = ids::id_generate_xbit(10);
        self.put_send_sensor_result_map(msg_id.clone(), temp_send_result);
        let res = self.easy_instance.send_sensor(
            remote_client_id,
            msg_id.clone(),
            sensor_event,
            sensor_data,
        );
        if res {
            return Ok(msg_id.to_owned());
        } else {
            let err_msg = "send_sensor transport error.".to_string();
            self.drain_send_sensor_result_callback(false, msg_id.to_owned(), err_msg.clone());
            anyhow::bail!(err_msg);
        }
    }

    fn send_bytebuf(
        &self,
        temp_send_result: Arc<Box<TempSendByteBufResultCallback>>, // 回调指针
        remote_client_id: String,                                  // 对端设备ID
        can_lost: bool,                                            // 能否丢包
        video_buf: Option<Vec<u8>>,                                // 视频数据
        audio_buf: Option<Vec<u8>>,                                // 音频数据
    ) -> Result<String, anyhow::Error> {
        let msg_id = ids::id_generate_xbit(10);
        self.put_send_bytebuf_result_map(msg_id.clone(), temp_send_result);
        let res = self.easy_instance.send_bytebuf(
            remote_client_id,
            msg_id.clone(),
            can_lost,
            video_buf,
            audio_buf,
        );
        if res {
            return Ok(msg_id.to_owned());
        } else {
            let err_msg = "send_bytebuf transport error.".to_string();
            self.drain_send_bytebuf_result_callback(false, msg_id.to_owned(), err_msg.clone());
            anyhow::bail!(err_msg);
        }
    }

    fn generate_p2prpc_file_mapping(
        &self,
        file_path: String, // 本地文件的绝对路径
    ) -> Result<String, anyhow::Error> {
        self.easy_instance.generate_p2prpc_file_mapping(file_path)
    }

    fn get_p2prpc_file_mapping_to_http(
        &self,
        remote_client_id: String, // 对方的设备id
        mapping_url: String,      // 映射的url名字
    ) -> Result<String, anyhow::Error> {
        self.easy_instance
            .get_p2prpc_file_mapping_to_http(remote_client_id, mapping_url)
    }

    fn query_online_from_gateway(
        &self,
        client_id: String,                                 // 接入的client_id
        query_callback: Arc<Box<TempQueryOnLineCallback>>, // 接口请求回调
    ) -> Result<(), anyhow::Error> {
        self.easy_instance
            .query_online_from_gateway(client_id, query_callback)
    }

    fn is_in_same_host(
        &self,
        remote_client_id: String, //远端client_id
    ) -> bool {
        self.easy_instance.is_in_same_host(remote_client_id)
    }
}

impl ICoolinkEasyApi for CoolinkSDKEasyApi {
    fn pre_init_sdk_log(log_init: LogIntArgs) {
        SGWebSocketContractImp::init_sg_sdk(log_init.tag, log_init.log_level);
    }

    fn init_easy_sdk(
        log_init: LogIntArgs,             // 日志初始化参数
        server_init: ServerQueryInitArgs, // 服务请求初始化参数
        httpd_init: HttpdInitArgs,        // httpd服务请求初始化参数
    ) {
        if set_easy_sdk_instance(CoolinkSDKEasyApi::new(log_init, server_init, httpd_init)) {
            log::info!("coolink easy api init!");
        } else {
            log::warn!("coolink easy api already inited!");
        }
    }

    /// 设置消息通道的回调
    fn set_message_type_tunnel_listener(listener: Box<dyn IUniqueMessageTunnelListener>) {
        match EASY_API_INSTANCE.try_get() {
            Some(instance) => {
                instance
                    .read()
                    .unwrap()
                    .set_message_type_tunnel_listener(listener);
            }
            None => {
                log::error!("error! please call init_easy_sdk first!");
            }
        };
    }

    /// 设置传感通道的回调
    fn set_sensor_type_tunnel_listener(listener: Box<dyn ISensorTunnelListener>) {
        match EASY_API_INSTANCE.try_get() {
            Some(instance) => {
                instance
                    .read()
                    .unwrap()
                    .set_sensor_type_tunnel_listener(listener);
            }
            None => {
                log::error!("error! please call init_easy_sdk first!");
            }
        };
    }

    /// 设置字节数组通道的回调
    fn set_bytebuf_type_tunnel_listener(listener: Box<dyn IByteBufListener>) {
        match EASY_API_INSTANCE.try_get() {
            Some(instance) => {
                instance
                    .read()
                    .unwrap()
                    .set_bytebuf_type_tunnel_listener(listener);
            }
            None => {
                log::error!("error! please call init_easy_sdk first!");
            }
        };
    }

    /// 重新设置用户，比如在切换账号之后
    fn re_init_user_id(user_id: String, // 新的用户id
    ) {
        match EASY_API_INSTANCE.try_get() {
            Some(instance) => {
                instance.read().unwrap().re_init_user_id(user_id);
            }
            None => {
                log::error!("error! please call init_easy_sdk first!");
            }
        };
    }

    /// 重新设置设备ID,比如在设备重新初始化之后
    fn re_init_device_id(device_id: String, // 新的用户id
    ) {
        match EASY_API_INSTANCE.try_get() {
            Some(instance) => {
                instance.read().unwrap().re_init_device_id(device_id);
            }
            None => {
                log::error!("error! please call init_easy_sdk first!");
            }
        };
    }

    /// 主动向对端设备发起消息通道的连接
    /// 返回int： 1 代表连接已经存在，默认是成功的，0代表此次成功，-1代表有错误发生
    fn connect_peer_as_controlling(remote_client_id: String, // 对端ID
    ) -> i32 {
        match EASY_API_INSTANCE.try_get() {
            Some(instance) => {
                return instance
                    .read()
                    .unwrap()
                    .connect_peer_as_controlling(remote_client_id);
            }
            None => {
                log::error!("error! please call init_easy_sdk first!");
                return -1;
            }
        };
    }

    /// 主动向对端设备发起sensor通道的连接
    /// 返回int： 1 代表连接已经存在，默认是成功的，0代表此次成功，-1代表有错误发生
    fn open_sensor_as_controlling(remote_client_id: String, // 对端ID
    ) -> i32 {
        match EASY_API_INSTANCE.try_get() {
            Some(instance) => {
                return instance
                    .read()
                    .unwrap()
                    .open_sensor_as_controlling(remote_client_id);
            }
            None => {
                log::error!("error! please call init_easy_sdk first!");
                return -1;
            }
        };
    }

    /// 主动向对端设备发起bytebuf通道的连接
    /// 返回int： 1 代表连接已经存在，默认是成功的，0代表此次成功，-1代表有错误发生
    fn start_byte_buf_channel_as_controlling(remote_client_id: String, // 对端ID
    ) -> i32 {
        match EASY_API_INSTANCE.try_get() {
            Some(instance) => {
                return instance
                    .read()
                    .unwrap()
                    .start_byte_buf_channel_as_controlling(remote_client_id);
            }
            None => {
                log::error!("error! please call init_easy_sdk first!");
                return -1;
            }
        };
    }

    /// 判断消息通道是否已经准备完成
    fn is_message_tunnel_ready(remote_client_id: String, // 对端ID
    ) -> bool {
        match EASY_API_INSTANCE.try_get() {
            Some(instance) => {
                return instance
                    .read()
                    .unwrap()
                    .is_message_tunnel_ready(remote_client_id);
            }
            None => {
                log::error!("error! please call init_easy_sdk first!");
                return false;
            }
        };
    }

    /// 判断sensor通道是否已经准备完成
    fn is_sensor_tunnel_ready(remote_client_id: String, // 对端ID
    ) -> bool {
        match EASY_API_INSTANCE.try_get() {
            Some(instance) => {
                return instance
                    .read()
                    .unwrap()
                    .is_sensor_tunnel_ready(remote_client_id);
            }
            None => {
                log::error!("error! please call init_easy_sdk first!");
                return false;
            }
        };
    }

    /// 判断buffer通道是否已经准备完成
    fn is_bytebuf_tunnel_ready(remote_client_id: String, // 对端ID
    ) -> bool {
        match EASY_API_INSTANCE.try_get() {
            Some(instance) => {
                return instance
                    .read()
                    .unwrap()
                    .is_bytebuf_tunnel_ready(remote_client_id);
            }
            None => {
                log::error!("error! please call init_easy_sdk first!");
                return false;
            }
        };
    }

    /// 主动端开通道
    fn destroy_tunnel(remote_client_id: String, // 对端ID
    ) -> bool {
        match EASY_API_INSTANCE.try_get() {
            Some(instance) => {
                return instance.read().unwrap().destroy_tunnel(remote_client_id);
            }
            None => {
                log::error!("error! please call init_easy_sdk first!");
                return false;
            }
        };
    }

    /// 获得当前创建了P2P通道的连接设备列表
    fn get_connected_devices() -> Option<Vec<Option<UniqueTunnel>>> {
        match EASY_API_INSTANCE.try_get() {
            Some(instance) => {
                return Some(instance.read().unwrap().get_connected_devices());
            }
            None => {
                log::error!("error! please call init_easy_sdk first!");
                return None;
            }
        };
    }

    /// 通过连接通道发送消息
    /// 该方法为异步方法，ISendMessageResult的message_id映射当前返回值
    /// 返回值代表了当前消息生成的message_id
    fn send_message(
        send_result: Box<dyn ISendMessageResult>, // 回调指针
        remote_client_id: String,                 // 对端设备ID
        message_event: String,                    // 消息名
        message: String,                          // 消息
    ) -> Result<String, anyhow::Error> {
        match EASY_API_INSTANCE.try_get() {
            Some(instance) => {
                let temp_send_cb = TempSendMsgResultCallback {
                    msg_send_result_listener: send_result,
                };

                return instance.read().unwrap().send_message(
                    Arc::new(Box::new(temp_send_cb)),
                    remote_client_id,
                    message_event,
                    message,
                );
            }
            None => {
                log::error!("error! please call init_easy_sdk first!");
                anyhow::bail!("error send_message but easy api not init.");
            }
        };
    }

    /// 通过连接通道发送广播消息，不需要带上remote_client_id参数
    /// 该方法为异步方法，ISendMessageResult的message_id映射当前返回值
    /// 返回值代表了一组(remote_client_id,message_id)的结果
    fn broadcast_message(
        send_result: Box<dyn ISendMessageResult>, //结果回调
        message_event: String,                    //消息名称
        message: String,                          // 消息内容
    ) -> Vec<Result<(String, String), anyhow::Error>> {
        match EASY_API_INSTANCE.try_get() {
            Some(instance) => {
                let temp_send_cb = TempSendMsgResultCallback {
                    msg_send_result_listener: send_result,
                };
                return instance.read().unwrap().broadcast_message(
                    Arc::new(Box::new(temp_send_cb)),
                    message_event,
                    message,
                );
            }
            None => {
                log::error!("error! please call init_easy_sdk first!");
                return vec![];
            }
        };
    }

    /// 通过sensor通道发送消息
    /// 该方法为异步方法，ISendSensorResult的message_id映射当前返回值
    /// 返回值代表了当前消息生成的message_id
    fn send_sensor(
        send_result: Box<dyn ISendSensorResult>, //回调结果
        remote_client_id: String,                // 对端设备ID
        sensor_event: String,                    // sensor消息名称
        sensor_data: String,                     // sensor消息
    ) -> Result<String, anyhow::Error> {
        match EASY_API_INSTANCE.try_get() {
            Some(instance) => {
                let temp_send_cb = TempSendSensorResultCallback {
                    sensor_send_result_listener: send_result,
                };
                return instance.read().unwrap().send_sensor(
                    Arc::new(Box::new(temp_send_cb)),
                    remote_client_id,
                    sensor_event,
                    sensor_data,
                );
            }
            None => {
                log::error!("error! please call init_easy_sdk first!");
                anyhow::bail!("error send_sensor but easy api not init.");
            }
        };
    }

    /// 通过流数组通道发送消息
    /// 该方法为异步方法，ISendSensorResult的message_id映射当前返回值
    /// 返回值代表了当前消息生成的message_id
    fn send_bytebuf(
        send_result: Box<dyn ISendByteBufResult>, //结果回调
        remote_client_id: String,                 // 对端设备ID
        can_lost: bool,                           // 能否丢包
        video_buf: Option<Vec<u8>>,               // 视频数据
        audio_buf: Option<Vec<u8>>,               // 音频数据
    ) -> Result<String, anyhow::Error> {
        match EASY_API_INSTANCE.try_get() {
            Some(instance) => {
                let temp_send_cb = TempSendByteBufResultCallback {
                    bytebuf_send_result_listener: send_result,
                };

                return instance.read().unwrap().send_bytebuf(
                    Arc::new(Box::new(temp_send_cb)),
                    remote_client_id,
                    can_lost,
                    video_buf,
                    audio_buf,
                );
            }
            None => {
                log::error!("error! please call init_easy_sdk first!");
                anyhow::bail!("error send_bytebuf but easy api not init.");
            }
        };
    }
    /// 将本地文件添加到P2P的mapping中
    fn generate_p2prpc_file_mapping(
        file_path: String, // 本地文件的绝对路径
    ) -> Result<String, anyhow::Error> {
        match EASY_API_INSTANCE.try_get() {
            Some(instance) => {
                return instance
                    .read()
                    .unwrap()
                    .generate_p2prpc_file_mapping(file_path);
            }
            None => {
                log::error!("error! please call init_easy_sdk first!");
                anyhow::bail!("error generate_p2prpc_file_mapping but easy api not init.");
            }
        };
    }

    /// 获取到对端法国来的mapping文件，生成url
    fn get_p2prpc_file_mapping_to_http(
        remote_client_id: String, // 对方的设备id
        mapping_url: String,      // 映射的url名字
    ) -> Result<String, anyhow::Error> {
        match EASY_API_INSTANCE.try_get() {
            Some(instance) => {
                return instance
                    .read()
                    .unwrap()
                    .get_p2prpc_file_mapping_to_http(remote_client_id, mapping_url);
            }
            None => {
                log::error!("error! please call init_easy_sdk first!");
                anyhow::bail!("error get_p2prpc_file_mapping_to_http but easy api not init.");
            }
        };
    }

    /// 从网关处请求判断client_id是否是在线状态
    fn query_online_from_gateway(
        client_id: String,                     // 接入的client_id
        query_callback: Box<dyn IQueryOnline>, // 接口请求回调
    ) -> Result<(), anyhow::Error> {
        match EASY_API_INSTANCE.try_get() {
            Some(instance) => {
                let temp_query_cb = TempQueryOnLineCallback {
                    temp_query_cb_listener: query_callback,
                };

                return instance
                    .read()
                    .unwrap()
                    .query_online_from_gateway(client_id, Arc::new(Box::new(temp_query_cb)));
            }
            None => {
                log::error!("error! please call init_easy_sdk first!");
                anyhow::bail!("error query_online_from_gateway but easy api not init.");
            }
        };
    }

    /// 判断是否在同一个host下
    fn is_in_same_host(remote_client_id: String, //远端client_id
    ) -> bool {
        match EASY_API_INSTANCE.try_get() {
            Some(instance) => {
                return instance.read().unwrap().is_in_same_host(remote_client_id);
            }
            None => {
                log::error!("error! please call init_easy_sdk first!");
                return false;
            }
        };
    }
}
